تحلیل فایل CSV با هوش مصنوعی؛ ساخت دستیار تحلیل داده از صفر

بخش بزرگی از اطلاعات کسب‌وکارها در فایل‌های CSV نگهداری می‌شود:

• خروجی فروشگاه اینترنتی
• گزارش فروش
• فهرست سفارش‌ها
• داده‌های مشتریان
• نتایج کمپین
• موجودی کالا
• گزارش پشتیبانی
• خروجی نرم‌افزار حسابداری
• داده‌های نظرسنجی
• گزارش رویدادهای محصول

کاربر معمولاً می‌خواهد پاسخ سؤال‌هایی مانند این‌ها را پیدا کند:

• فروش هر ماه چقدر بوده است؟
• کدام دسته محصول بیشترین درآمد را ساخته است؟
• متوسط مبلغ سفارش در هر شهر چقدر است؟
• چند درصد سفارش‌ها لغو شده‌اند؟
• کدام محصول در حال کاهش فروش است؟
• چه ستون‌هایی داده ناقص دارند؟
• آیا داده تکراری وجود دارد؟
• چه روند یا ناهنجاری قابل‌توجهی دیده می‌شود؟

مدل‌های هوش مصنوعی می‌توانند ساختار داده و هدف کاربر را درک کنند، اما نباید محاسبات عددی نهایی را فقط به مدل زبانی بسپاریم.

معماری قابل اعتماد چنین است:

1. فایل CSV با پایتون خوانده می‌شود.
2. نوع ستون‌ها و کیفیت داده بررسی می‌شود.
3. فقط خلاصه Schema و نمونه کنترل‌شده برای مدل ارسال می‌شود.
4. مدل یک Analysis Plan ساختاریافته تولید می‌کند.
5. برنامه در Backend اعتبارسنجی می‌شود.
6. محاسبات با Pandas انجام می‌شوند.
7. نتیجه قطعی برای مدل ارسال می‌شود.
8. مدل نتیجه را به زبان فارسی توضیح می‌دهد.

در این روش، مدل تصمیم می‌گیرد برای پاسخ به سؤال چه تحلیل‌هایی لازم است، اما جمع، میانگین، فیلتر و Group By را Pandas انجام می‌دهد.

تحلیل CSV با هوش مصنوعی چیست؟

AI CSV Analysis یعنی استفاده از مدل هوش مصنوعی برای تبدیل سؤال زبان طبیعی به یک جریان تحلیل داده.

ورودی:

فروش تکمیل‌شده هر شهر را محاسبه و سه شهر اول را نمایش بده.

برنامه تحلیلی:

{
  "filters": [
    {
      "column": "status",
      "operator": "eq",
      "value": "completed"
    }
  ],
  "group_by": [
    "city"
  ],
  "aggregations": [
    {
      "column": "total_amount",
      "function": "sum",
      "alias": "total_sales"
    }
  ],
  "sort": [
    {
      "column": "total_sales",
      "direction": "desc"
    }
  ],
  "limit": 3
}

نتیجه توسط Pandas محاسبه می‌شود. سپس مدل فقط توضیح می‌دهد:

تهران با فروش ۱۲۵ میلیون ریال در رتبه اول قرار دارد و پس از آن
شیراز و اصفهان قرار گرفته‌اند.

چرا نباید فایل کامل را مستقیماً برای مدل ارسال کنیم؟

محدودیت Context

یک فایل چندصد مگابایتی در Context مدل جا نمی‌شود.

هزینه

ارسال هزاران ردیف مشابه هزینه غیرضروری ایجاد می‌کند.

دقت محاسبات

مدل زبانی ماشین‌حساب یا موتور تحلیل ستونی نیست. ممکن است مجموع، درصد یا میانگین را اشتباه محاسبه کند.

داده‌های غیرضروری

برای پاسخ به «فروش هر ماه» نیازی به ارسال نام و اطلاعات تمام مشتریان نیست.

تکرارپذیری

تحلیل Pandas یا SQL قابل اجرا و بازتولید است؛ پاسخ محاسباتی مستقیم مدل ممکن است در اجراهای مختلف تفاوت داشته باشد.

اعتبارسنجی

یک Analysis Plan JSON را می‌توان پیش از اجرا بررسی کرد. متن آزاد چنین قابلیتی ندارد.

AI و Pandas چه نقشی دارند؟

کاربردهای عملی

تحلیل فروش

• فروش روزانه و ماهانه
• محصول پرفروش
• متوسط ارزش سفارش
• نرخ لغو
• فروش هر شهر
• مقایسه دسته‌ها

تحلیل مشتری

• تعداد مشتری فعال
• خرید تکراری
• متوسط خرید
• مشتریان با بیشترین ارزش
• فاصله میان خریدها

تحلیل کمپین

• نرخ کلیک
• نرخ تبدیل
• هزینه هر تبدیل
• مقایسه کانال‌ها
• عملکرد زمانی کمپین

تحلیل پشتیبانی

• تعداد تیکت‌ها
• زمان متوسط پاسخ
• موضوعات پرتکرار
• روند رضایت
• حجم تیکت هر تیم

کنترل کیفیت داده

• مقدارهای خالی
• رکوردهای تکراری
• نوع داده ناسازگار
• مقدارهای خارج از دامنه
• ستون‌های تقریباً ثابت
• تاریخ‌های نامعتبر

چالش‌های CSV فارسی

Encoding

فایل ممکن است یکی از Encodingهای زیر را داشته باشد:

• UTF-8
• UTF-8 with BOM
• Windows-1256
• UTF-16

جداکننده

همه فایل‌ها از Comma استفاده نمی‌کنند:

,
;
\t

ارقام فارسی و عربی

۱۲۵۰۰۰
١٢٥٠٠٠
125000

جداکننده هزارگان

1,250,000
۱٬۲۵۰٬۰۰۰
۱،۲۵۰،۰۰۰

تاریخ شمسی

۱۴۰۵/۰۴/۲۹

Pandas به‌صورت پیش‌فرض این تاریخ را مانند تاریخ میلادی تحلیل نمی‌کند.

ریال و تومان

واحد مبلغ باید در Metadata مشخص باشد. مدل نباید از بزرگی عدد واحد را حدس بزند.

نام ستون‌های فارسی

Pandas از نام فارسی پشتیبانی می‌کند، اما برای API داخلی می‌توان Mapping کنترل‌شده ساخت:

{
  "تاریخ سفارش": "order_date",
  "مبلغ نهایی": "total_amount",
  "وضعیت": "status"
}

نام اصلی باید برای نمایش حفظ شود.

پروژه عملی این آموزش

یک API می‌سازیم که:

• CSV را دریافت می‌کند.
• Encoding و جداکننده را تشخیص می‌دهد.
• داده را پروفایل می‌کند.
• فایل را با یک شناسه موقت نگه می‌دارد.
• سؤال فارسی دریافت می‌کند.
• سؤال را با API درواره به Analysis Plan تبدیل می‌کند.
• فقط عملیات مجاز را اجرا می‌کند.
• نتیجه جدول و خلاصه فارسی برمی‌گرداند.

مدل اجازه تولید و اجرای Python دلخواه ندارد.

ایجاد پروژه

mkdir ai-csv-analyst
cd ai-csv-analyst

python -m venv .venv

فعال‌سازی در Linux و macOS:

source .venv/bin/activate

فعال‌سازی در Windows:

.venv\Scripts\Activate.ps1

نصب وابستگی‌ها:

pip install \
  openai \
  python-dotenv \
  pydantic \
  pandas \
  fastapi \
  uvicorn \
  python-multipart \
  charset-normalizer

ساخت پوشه‌ها:

mkdir analyst data output

فایل‌های زیر را ایجاد کنید:

analyst/__init__.py
analyst/schemas.py
analyst/csv_loader.py
analyst/profiler.py
analyst/planner.py
analyst/validator.py
analyst/executor.py
analyst/explainer.py
analyst/storage.py
main.py

تنظیم API درواره

فایل .env:

DARVAREH_API_KEY=YOUR_API_KEY
DARVAREH_MODEL=MODEL_ID_DARVAREH

فایل .gitignore:

.env
.venv/
__pycache__/
data/
output/

برای دریافت API Key در درواره ثبت‌نام کنید. مدل مناسب را از صفحه مدل‌های درواره انتخاب کنید.

ساخت CSV آزمایشی

فایل sales.csv:

order_id,order_date,customer_id,city,category,product,quantity,unit_price,total_amount,status
1001,2026-06-01,C-01,تهران,لوازم جانبی,ماوس بی‌سیم,2,1200000,2400000,completed
1002,2026-06-03,C-02,شیراز,مانیتور,مانیتور 27 اینچ,1,18500000,18500000,completed
1003,2026-06-05,C-03,تهران,لپ‌تاپ,لپ‌تاپ مدل A,1,52000000,52000000,cancelled
1004,2026-06-12,C-01,تهران,لوازم جانبی,کیبورد مکانیکی,1,3500000,3500000,completed
1005,2026-07-02,C-04,اصفهان,لپ‌تاپ,لپ‌تاپ مدل A,1,52000000,52000000,completed
1006,2026-07-04,C-02,شیراز,لوازم جانبی,هاب USB-C,2,2100000,4200000,completed
1007,2026-07-10,C-05,تهران,مانیتور,مانیتور 27 اینچ,1,18500000,18500000,pending
1008,2026-07-14,C-01,تهران,لوازم جانبی,ماوس بی‌سیم,3,1200000,3600000,completed

تعریف Schema برنامه تحلیل

فایل analyst/schemas.py:

from typing import (
    Any,
    Literal,
)

from pydantic import BaseModel, Field


FilterOperator = Literal[
    "eq",
    "ne",
    "gt",
    "gte",
    "lt",
    "lte",
    "contains",
    "in",
    "is_null",
    "not_null",
]

AggregationFunction = Literal[
    "sum",
    "mean",
    "median",
    "min",
    "max",
    "count",
    "nunique",
]

SortDirection = Literal[
    "asc",
    "desc",
]


class FilterRule(BaseModel):
    column: str
    operator: FilterOperator
    value: Any | None = None


class AggregationRule(BaseModel):
    column: str
    function: AggregationFunction
    alias: str


class SortRule(BaseModel):
    column: str
    direction: SortDirection


class ComputedMetric(BaseModel):
    name: str
    numerator_column: str
    denominator_column: str
    multiplier: float = 1.0


class AnalysisPlan(BaseModel):
    question_interpretation: str
    clarification_needed: bool = False
    clarification_question: str | None = None
    selected_columns: list[str] = Field(
        default_factory=list
    )
    filters: list[FilterRule] = Field(
        default_factory=list
    )
    group_by: list[str] = Field(
        default_factory=list
    )
    aggregations: list[
        AggregationRule
    ] = Field(default_factory=list)
    sort: list[SortRule] = Field(
        default_factory=list
    )
    limit: int = Field(
        default=100,
        ge=1,
        le=1000,
    )
    assumptions: list[str] = Field(
        default_factory=list
    )


class DatasetColumn(BaseModel):
    name: str
    dtype: str
    non_null_count: int
    null_count: int
    unique_count: int
    sample_values: list[Any] = Field(
        default_factory=list
    )
    minimum: Any | None = None
    maximum: Any | None = None
    mean: float | None = None


class DatasetProfile(BaseModel):
    row_count: int
    column_count: int
    columns: list[DatasetColumn]
    duplicate_rows: int
    memory_bytes: int
    warnings: list[str] = Field(
        default_factory=list
    )


class AnalysisRequest(BaseModel):
    dataset_id: str
    question: str


class AnalysisResult(BaseModel):
    dataset_id: str
    question: str
    plan: AnalysisPlan
    columns: list[str] = Field(
        default_factory=list
    )
    rows: list[list[Any]] = Field(
        default_factory=list
    )
    row_count: int = 0
    explanation: str | None = None
    warnings: list[str] = Field(
        default_factory=list
    )

خواندن CSV با کنترل Encoding و Separator

فایل analyst/csv_loader.py:

import csv
import io

import pandas as pd
from charset_normalizer import (
    from_bytes,
)


class CSVLoadError(ValueError):
    pass


MAX_ROWS = 500_000
MAX_COLUMNS = 200


def detect_encoding(
    file_bytes: bytes,
) -> str:
    match = from_bytes(
        file_bytes[:200_000]
    ).best()

    if match is None:
        return "utf-8"

    return match.encoding or "utf-8"


def detect_delimiter(
    text_sample: str,
) -> str:
    try:
        dialect = csv.Sniffer().sniff(
            text_sample,
            delimiters=",;\t|",
        )

        return dialect.delimiter
    except csv.Error:
        return ","


def load_csv(
    file_bytes: bytes,
) -> tuple[pd.DataFrame, dict]:
    if not file_bytes:
        raise CSVLoadError(
            "CSV file is empty."
        )

    encoding = detect_encoding(
        file_bytes
    )

    try:
        text = file_bytes.decode(
            encoding,
            errors="strict",
        )
    except UnicodeDecodeError as error:
        raise CSVLoadError(
            f"Could not decode CSV with "
            f"encoding {encoding}."
        ) from error

    delimiter = detect_delimiter(
        text[:20_000]
    )

    try:
        dataframe = pd.read_csv(
            io.StringIO(text),
            sep=delimiter,
            nrows=MAX_ROWS + 1,
        )
    except Exception as error:
        raise CSVLoadError(
            f"Could not parse CSV: {error}"
        ) from error

    if len(dataframe) > MAX_ROWS:
        raise CSVLoadError(
            f"CSV contains more than "
            f"{MAX_ROWS} rows."
        )

    if len(dataframe.columns) > MAX_COLUMNS:
        raise CSVLoadError(
            f"CSV contains more than "
            f"{MAX_COLUMNS} columns."
        )

    dataframe.columns = [
        str(column).strip()
        for column in dataframe.columns
    ]

    if len(set(dataframe.columns)) != len(
        dataframe.columns
    ):
        raise CSVLoadError(
            "CSV contains duplicate "
            "column names."
        )

    metadata = {
        "encoding": encoding,
        "delimiter": delimiter,
        "rows": len(dataframe),
        "columns": len(
            dataframe.columns
        ),
    }

    return dataframe, metadata

برای فایل بسیار بزرگ بهتر است از Chunk Processing، دیتابیس تحلیلی یا موتورهایی مانند DuckDB استفاده شود.

نرمال‌سازی اعداد فارسی

فایل analyst/normalization.py:

import re

import pandas as pd


PERSIAN_DIGITS = "۰۱۲۳۴۵۶۷۸۹"
ARABIC_DIGITS = "٠١٢٣٤٥٦٧٨٩"
LATIN_DIGITS = "0123456789"

DIGIT_TRANSLATION = str.maketrans(
    PERSIAN_DIGITS + ARABIC_DIGITS,
    LATIN_DIGITS + LATIN_DIGITS,
)


def normalize_digit_string(
    value: str,
) -> str:
    return value.translate(
        DIGIT_TRANSLATION
    )


def normalize_numeric_text(
    value: str,
) -> str:
    normalized = normalize_digit_string(
        value
    )

    normalized = re.sub(
        r"[,،٬\s]",
        "",
        normalized,
    )

    return normalized


def infer_numeric_column(
    series: pd.Series,
    minimum_success_rate: float = 0.95,
) -> pd.Series | None:
    if not (
        pd.api.types
        .is_object_dtype(series)
    ):
        return None

    non_null = series.dropna()

    if non_null.empty:
        return None

    normalized = non_null.astype(
        str
    ).map(normalize_numeric_text)

    converted = pd.to_numeric(
        normalized,
        errors="coerce",
    )

    success_rate = (
        converted.notna().mean()
    )

    if success_rate < minimum_success_rate:
        return None

    full_normalized = (
        series.astype("string")
        .map(
            lambda value: (
                normalize_numeric_text(value)
                if pd.notna(value)
                else value
            )
        )
    )

    return pd.to_numeric(
        full_normalized,
        errors="coerce",
    )

تبدیل نوع نباید کورکورانه انجام شود. ستون‌هایی مانند شماره سفارش ممکن است عددی به نظر برسند اما Metric قابل جمع‌زدن نیستند.

پروفایل‌کردن Dataset

فایل analyst/profiler.py:

from typing import Any

import pandas as pd

from analyst.schemas import (
    DatasetColumn,
    DatasetProfile,
)


MAX_SAMPLE_VALUES = 5


def make_json_safe(
    value: Any,
) -> Any:
    if pd.isna(value):
        return None

    if hasattr(value, "isoformat"):
        return value.isoformat()

    if hasattr(value, "item"):
        return value.item()

    return value


def profile_dataframe(
    dataframe: pd.DataFrame,
) -> DatasetProfile:
    columns: list[DatasetColumn] = []
    warnings: list[str] = []

    for column_name in (
        dataframe.columns
    ):
        series = dataframe[column_name]
        non_null = series.dropna()

        samples = [
            make_json_safe(value)
            for value in (
                non_null.drop_duplicates()
                .head(MAX_SAMPLE_VALUES)
                .tolist()
            )
        ]

        minimum = None
        maximum = None
        mean = None

        if (
            pd.api.types
            .is_numeric_dtype(series)
            and not non_null.empty
        ):
            minimum = make_json_safe(
                non_null.min()
            )
            maximum = make_json_safe(
                non_null.max()
            )
            mean = float(
                non_null.mean()
            )

        null_count = int(
            series.isna().sum()
        )

        if null_count > 0:
            null_rate = (
                null_count
                / len(dataframe)
                if len(dataframe)
                else 0
            )

            if null_rate >= 0.25:
                warnings.append(
                    f"ستون {column_name} دارای "
                    f"{null_rate:.1%} مقدار خالی است."
                )

        columns.append(
            DatasetColumn(
                name=str(column_name),
                dtype=str(series.dtype),
                non_null_count=int(
                    series.notna().sum()
                ),
                null_count=null_count,
                unique_count=int(
                    series.nunique(
                        dropna=True
                    )
                ),
                sample_values=samples,
                minimum=minimum,
                maximum=maximum,
                mean=mean,
            )
        )

    duplicate_rows = int(
        dataframe.duplicated().sum()
    )

    if duplicate_rows:
        warnings.append(
            f"{duplicate_rows} ردیف کاملاً "
            "تکراری شناسایی شد."
        )

    return DatasetProfile(
        row_count=len(dataframe),
        column_count=len(
            dataframe.columns
        ),
        columns=columns,
        duplicate_rows=duplicate_rows,
        memory_bytes=int(
            dataframe.memory_usage(
                deep=True
            ).sum()
        ),
        warnings=warnings,
    )

برای مدل فقط Profile و نمونه مقدارهای محدود ارسال می‌شود، نه کل Dataset.

نگهداری موقت Dataset

فایل analyst/storage.py:

import secrets
from dataclasses import dataclass
from datetime import (
    datetime,
    timedelta,
    timezone,
)

import pandas as pd


DATASET_TTL = timedelta(hours=1)


@dataclass
class StoredDataset:
    dataframe: pd.DataFrame
    created_at: datetime


class DatasetStore:
    def __init__(self):
        self._datasets: dict[
            str,
            StoredDataset,
        ] = {}

    def save(
        self,
        dataframe: pd.DataFrame,
    ) -> str:
        self.cleanup()

        dataset_id = secrets.token_urlsafe(
            16
        )

        self._datasets[dataset_id] = (
            StoredDataset(
                dataframe=dataframe,
                created_at=datetime.now(
                    timezone.utc
                ),
            )
        )

        return dataset_id

    def get(
        self,
        dataset_id: str,
    ) -> pd.DataFrame | None:
        self.cleanup()

        stored = self._datasets.get(
            dataset_id
        )

        if stored is None:
            return None

        return stored.dataframe.copy()

    def cleanup(self) -> None:
        now = datetime.now(
            timezone.utc
        )

        expired_ids = [
            dataset_id
            for dataset_id, stored
            in self._datasets.items()
            if (
                now - stored.created_at
                > DATASET_TTL
            )
        ]

        for dataset_id in expired_ids:
            del self._datasets[
                dataset_id
            ]


dataset_store = DatasetStore()

این حافظه برای نمونه آموزشی است. در Production باید Storage و Lifecycle مناسب طراحی شود.

تولید Analysis Plan با API درواره

فایل analyst/planner.py:

import json
import os

from dotenv import load_dotenv
from openai import OpenAI
from pydantic import ValidationError

from analyst.schemas import (
    AnalysisPlan,
    DatasetProfile,
)


load_dotenv()

api_key = os.getenv("DARVAREH_API_KEY")
model = os.getenv("DARVAREH_MODEL")

if not api_key:
    raise RuntimeError(
        "DARVAREH_API_KEY is not configured."
    )

if not model:
    raise RuntimeError(
        "DARVAREH_MODEL is not configured."
    )

client = OpenAI(
    api_key=api_key,
    base_url="https://api.darvareh.ir/v1",
)


SYSTEM_PROMPT = """
تو یک تحلیلگر داده هستی.

وظیفه:
سؤال فارسی کاربر را بر اساس Dataset Profile به یک
Analysis Plan ساختاریافته تبدیل کن.

عملیات مجاز:
- انتخاب ستون
- Filter
- Group By
- Aggregation
- Sort
- Limit

قواعد:
- فقط از ستون‌های موجود استفاده کن.
- مقدار یا ستون جدید اختراع نکن.
- اگر سؤال مبهم است، clarification_needed را true کن.
- محاسبات را خودت انجام نده.
- نتیجه را حدس نزن.
- برای سؤال‌های فهرستی Limit قرار بده.
- شناسه‌ها را Sum یا Mean نکن.
- برای فروش فقط در صورت تعریف صریح، وضعیت مناسب را Filter کن.
- اگر واحد مبلغ مشخص نیست، آن را حدس نزن.
- خروجی فقط JSON معتبر باشد.
"""


OUTPUT_TEMPLATE = {
    "question_interpretation": "string",
    "clarification_needed": False,
    "clarification_question": None,
    "selected_columns": [],
    "filters": [
        {
            "column": "status",
            "operator": "eq",
            "value": "completed",
        }
    ],
    "group_by": [
        "city"
    ],
    "aggregations": [
        {
            "column": "total_amount",
            "function": "sum",
            "alias": "total_sales",
        }
    ],
    "sort": [
        {
            "column": "total_sales",
            "direction": "desc",
        }
    ],
    "limit": 10,
    "assumptions": [],
}


def generate_analysis_plan(
    question: str,
    profile: DatasetProfile,
    dataset_notes: list[str],
) -> AnalysisPlan:
    payload = {
        "question": question,
        "dataset_profile": (
            profile.model_dump()
        ),
        "dataset_notes": dataset_notes,
    }

    response = client.chat.completions.create(
        model=model,
        temperature=0.1,
        messages=[
            {
                "role": "system",
                "content": SYSTEM_PROMPT,
            },
            {
                "role": "user",
                "content": (
                    "برای داده زیر Plan بساز:\n\n"
                    + json.dumps(
                        payload,
                        ensure_ascii=False,
                        indent=2,
                    )
                    + "\n\nقالب خروجی:\n"
                    + json.dumps(
                        OUTPUT_TEMPLATE,
                        ensure_ascii=False,
                        indent=2,
                    )
                ),
            },
        ],
    )

    raw_output = (
        response.choices[0]
        .message.content
    )

    if not raw_output:
        raise RuntimeError(
            "The model returned an "
            "empty plan."
        )

    try:
        parsed = json.loads(raw_output)
    except json.JSONDecodeError as error:
        raise RuntimeError(
            f"Invalid JSON from model: "
            f"{error}"
        ) from error

    try:
        return AnalysisPlan.model_validate(
            parsed
        )
    except ValidationError as error:
        raise RuntimeError(
            f"Analysis plan validation "
            f"failed: {error}"
        ) from error

یادداشت‌های معنایی Dataset

Profile فقط ساختار را نشان می‌دهد. قواعد کسب‌وکار باید جداگانه ثبت شوند:

DATASET_NOTES = [
    "هر ردیف یک سفارش است.",
    "total_amount مبلغ نهایی سفارش به ریال است.",
    "سفارش موفق یعنی status برابر completed.",
    "سفارش cancelled نباید در فروش نهایی محاسبه شود.",
    "order_date تاریخ میلادی با فرمت YYYY-MM-DD است.",
    "order_id و customer_id شناسه‌اند و Metric عددی نیستند.",
]

بدون این توضیحات، مدل نمی‌داند «فروش» فقط شامل سفارش‌های تکمیل‌شده است.

اعتبارسنجی Plan

فایل analyst/validator.py:

import pandas as pd

from analyst.schemas import (
    AnalysisPlan,
)


class PlanValidationError(
    ValueError
):
    pass


def validate_plan(
    plan: AnalysisPlan,
    dataframe: pd.DataFrame,
) -> None:
    available_columns = set(
        map(str, dataframe.columns)
    )

    referenced_columns = set(
        plan.selected_columns
    )

    referenced_columns.update(
        rule.column
        for rule in plan.filters
    )

    referenced_columns.update(
        plan.group_by
    )

    referenced_columns.update(
        rule.column
        for rule in plan.aggregations
    )

    unknown_columns = (
        referenced_columns
        - available_columns
    )

    if unknown_columns:
        raise PlanValidationError(
            "Unknown columns: "
            + ", ".join(
                sorted(unknown_columns)
            )
        )

    aliases = {
        rule.alias
        for rule in plan.aggregations
    }

    allowed_sort_columns = (
        available_columns
        | aliases
        | set(plan.group_by)
    )

    invalid_sort_columns = {
        rule.column
        for rule in plan.sort
        if (
            rule.column
            not in allowed_sort_columns
        )
    }

    if invalid_sort_columns:
        raise PlanValidationError(
            "Invalid sort columns: "
            + ", ".join(
                sorted(
                    invalid_sort_columns
                )
            )
        )

    for aggregation in (
        plan.aggregations
    ):
        series = dataframe[
            aggregation.column
        ]

        if (
            aggregation.function
            in {
                "sum",
                "mean",
                "median",
            }
            and not (
                pd.api.types
                .is_numeric_dtype(series)
            )
        ):
            raise PlanValidationError(
                f"Aggregation "
                f"{aggregation.function} "
                f"requires numeric column: "
                f"{aggregation.column}"
            )

    if (
        len(aliases)
        != len(plan.aggregations)
    ):
        raise PlanValidationError(
            "Aggregation aliases "
            "must be unique."
        )

اجرای فیلترها

فایل analyst/executor.py:

from typing import Any

import pandas as pd

from analyst.schemas import (
    AnalysisPlan,
    FilterRule,
)


class AnalysisExecutionError(
    RuntimeError
):
    pass


def apply_filter(
    dataframe: pd.DataFrame,
    rule: FilterRule,
) -> pd.DataFrame:
    series = dataframe[rule.column]
    operator = rule.operator
    value = rule.value

    if operator == "eq":
        mask = series == value
    elif operator == "ne":
        mask = series != value
    elif operator == "gt":
        mask = series > value
    elif operator == "gte":
        mask = series >= value
    elif operator == "lt":
        mask = series < value
    elif operator == "lte":
        mask = series <= value
    elif operator == "contains":
        mask = (
            series.astype("string")
            .str.contains(
                str(value),
                case=False,
                na=False,
                regex=False,
            )
        )
    elif operator == "in":
        if not isinstance(value, list):
            raise AnalysisExecutionError(
                "The 'in' operator "
                "requires a list."
            )

        mask = series.isin(value)
    elif operator == "is_null":
        mask = series.isna()
    elif operator == "not_null":
        mask = series.notna()
    else:
        raise AnalysisExecutionError(
            f"Unsupported operator: "
            f"{operator}"
        )

    return dataframe.loc[mask]


def execute_plan(
    dataframe: pd.DataFrame,
    plan: AnalysisPlan,
) -> pd.DataFrame:
    result = dataframe.copy()

    for filter_rule in plan.filters:
        result = apply_filter(
            result,
            filter_rule,
        )

    if plan.aggregations:
        named_aggregations = {
            aggregation.alias: pd.NamedAgg(
                column=aggregation.column,
                aggfunc=aggregation.function,
            )
            for aggregation
            in plan.aggregations
        }

        if plan.group_by:
            result = (
                result.groupby(
                    plan.group_by,
                    dropna=False,
                )
                .agg(**named_aggregations)
                .reset_index()
            )
        else:
            values: dict[str, Any] = {}

            for aggregation in (
                plan.aggregations
            ):
                series = result[
                    aggregation.column
                ]

                function = getattr(
                    series,
                    aggregation.function,
                )

                values[
                    aggregation.alias
                ] = function()

            result = pd.DataFrame(
                [values]
            )

    elif plan.selected_columns:
        result = result[
            plan.selected_columns
        ]

    for sort_rule in reversed(
        plan.sort
    ):
        result = result.sort_values(
            by=sort_rule.column,
            ascending=(
                sort_rule.direction
                == "asc"
            ),
            kind="stable",
            na_position="last",
        )

    result = result.head(plan.limit)

    return result

هیچ Python تولیدشده توسط مدل اجرا نمی‌شود. مدل فقط یک Plan محدود و قابل اعتبارسنجی می‌سازد.

تبدیل نتیجه به JSON امن

مقادیر NumPy و Timestamp باید نرمال شوند:

from typing import Any

import pandas as pd


def json_safe_value(
    value: Any,
) -> Any:
    if pd.isna(value):
        return None

    if hasattr(value, "isoformat"):
        return value.isoformat()

    if hasattr(value, "item"):
        return value.item()

    return value


def dataframe_to_table(
    dataframe: pd.DataFrame,
) -> tuple[list[str], list[list]]:
    columns = [
        str(column)
        for column in dataframe.columns
    ]

    rows = [
        [
            json_safe_value(value)
            for value in row
        ]
        for row in dataframe.itertuples(
            index=False,
            name=None,
        )
    ]

    return columns, rows

توضیح نتیجه با مدل

فایل analyst/explainer.py:

import json

from analyst.planner import (
    client,
    model,
)
from analyst.schemas import (
    AnalysisPlan,
)


SYSTEM_PROMPT = """
تو نتیجه قطعی تحلیل داده را به زبان فارسی توضیح می‌دهی.

قواعد:
- فقط از جدول و Metadata ورودی استفاده کن.
- عدد جدید محاسبه یا اختراع نکن.
- واحد را فقط در صورت وجود در Metadata بنویس.
- Assumptionها را صریح بیان کن.
- میان نتیجه و تفسیر تفاوت بگذار.
- اگر نتیجه خالی است، آن را اعلام کن.
- پاسخ کوتاه، دقیق و مدیریتی باشد.
"""


def explain_analysis(
    question: str,
    plan: AnalysisPlan,
    columns: list[str],
    rows: list[list],
    dataset_notes: list[str],
) -> str:
    payload = {
        "question": question,
        "plan": plan.model_dump(),
        "result": {
            "columns": columns,
            "rows": rows[:100],
            "row_count": len(rows),
        },
        "dataset_notes": dataset_notes,
    }

    response = client.chat.completions.create(
        model=model,
        temperature=0.1,
        messages=[
            {
                "role": "system",
                "content": SYSTEM_PROMPT,
            },
            {
                "role": "user",
                "content": json.dumps(
                    payload,
                    ensure_ascii=False,
                    indent=2,
                ),
            },
        ],
    )

    explanation = (
        response.choices[0]
        .message.content
    )

    if not explanation:
        raise RuntimeError(
            "The model returned an "
            "empty explanation."
        )

    return explanation

برای Dataset بزرگ فقط نتیجه Aggregated یا تعداد محدودی ردیف ارسال کنید.

ساخت API با FastAPI

فایل main.py:

from fastapi import (
    FastAPI,
    File,
    HTTPException,
    UploadFile,
)

from analyst.csv_loader import (
    CSVLoadError,
    load_csv,
)
from analyst.executor import (
    AnalysisExecutionError,
    execute_plan,
)
from analyst.explainer import (
    explain_analysis,
)
from analyst.planner import (
    generate_analysis_plan,
)
from analyst.profiler import (
    profile_dataframe,
)
from analyst.schemas import (
    AnalysisRequest,
    AnalysisResult,
)
from analyst.storage import (
    dataset_store,
)
from analyst.validator import (
    PlanValidationError,
    validate_plan,
)


MAX_UPLOAD_SIZE = 25 * 1024 * 1024

DATASET_NOTES = [
    "هر ردیف یک سفارش است.",
    "total_amount مبلغ نهایی سفارش به ریال است.",
    "سفارش موفق یعنی status برابر completed.",
    "سفارش cancelled در فروش نهایی محاسبه نمی‌شود.",
    "order_id و customer_id شناسه هستند.",
]

app = FastAPI(
    title="Darvareh AI CSV Analyst",
    version="1.0.0",
)


@app.get("/health")
def health_check():
    return {
        "status": "ok",
    }


@app.post("/datasets")
async def upload_dataset(
    file: UploadFile = File(...),
):
    if not file.filename.lower().endswith(
        ".csv"
    ):
        raise HTTPException(
            status_code=415,
            detail="Only CSV files are supported.",
        )

    file_bytes = await file.read()

    if len(file_bytes) > MAX_UPLOAD_SIZE:
        raise HTTPException(
            status_code=413,
            detail="CSV file is too large.",
        )

    try:
        dataframe, metadata = load_csv(
            file_bytes
        )

        profile = profile_dataframe(
            dataframe
        )

        dataset_id = dataset_store.save(
            dataframe
        )

        return {
            "dataset_id": dataset_id,
            "filename": file.filename,
            "metadata": metadata,
            "profile": profile.model_dump(),
        }

    except CSVLoadError as error:
        raise HTTPException(
            status_code=400,
            detail=str(error),
        ) from error


@app.post(
    "/analyze",
    response_model=AnalysisResult,
)
def analyze_dataset(
    request: AnalysisRequest,
):
    dataframe = dataset_store.get(
        request.dataset_id
    )

    if dataframe is None:
        raise HTTPException(
            status_code=404,
            detail=(
                "Dataset not found or expired."
            ),
        )

    question = request.question.strip()

    if not question:
        raise HTTPException(
            status_code=422,
            detail="Question cannot be empty.",
        )

    if len(question) > 1000:
        raise HTTPException(
            status_code=422,
            detail="Question is too long.",
        )

    try:
        profile = profile_dataframe(
            dataframe
        )

        plan = generate_analysis_plan(
            question,
            profile,
            DATASET_NOTES,
        )

        if plan.clarification_needed:
            return AnalysisResult(
                dataset_id=request.dataset_id,
                question=question,
                plan=plan,
                warnings=[
                    plan.clarification_question
                    or "Clarification is required."
                ],
            )

        validate_plan(
            plan,
            dataframe,
        )

        result_dataframe = execute_plan(
            dataframe,
            plan,
        )

        columns, rows = dataframe_to_table(
            result_dataframe
        )

        explanation = explain_analysis(
            question,
            plan,
            columns,
            rows,
            DATASET_NOTES,
        )

        return AnalysisResult(
            dataset_id=request.dataset_id,
            question=question,
            plan=plan,
            columns=columns,
            rows=rows,
            row_count=len(rows),
            explanation=explanation,
            warnings=profile.warnings,
        )

    except (
        PlanValidationError,
        AnalysisExecutionError,
    ) as error:
        raise HTTPException(
            status_code=422,
            detail=str(error),
        ) from error

    except Exception as error:
        raise HTTPException(
            status_code=500,
            detail="Analysis failed.",
        ) from error

تابع dataframe_to_table را از ماژولی جدا Import کنید یا همان تابع بخش قبل را در analyst/executor.py قرار دهید.

اجرای API

uvicorn main:app --reload

مستندات تعاملی:

http://127.0.0.1:8000/docs

بارگذاری CSV با curl

curl -X POST \
  http://127.0.0.1:8000/datasets \
  -H "accept: application/json" \
  -H "Content-Type: multipart/form-data" \
  -F "file=@sales.csv"

پاسخ:

{
  "dataset_id": "DATASET_ID",
  "filename": "sales.csv",
  "metadata": {
    "encoding": "utf_8",
    "delimiter": ",",
    "rows": 8,
    "columns": 10
  },
  "profile": {
    "row_count": 8,
    "column_count": 10,
    "duplicate_rows": 0,
    "warnings": []
  }
}

پرسیدن سؤال فارسی

curl -X POST \
  http://127.0.0.1:8000/analyze \
  -H "Content-Type: application/json" \
  -d '{
    "dataset_id": "DATASET_ID",
    "question": "فروش سفارش‌های تکمیل‌شده در هر شهر را محاسبه و از بیشترین به کمترین مرتب کن"
  }'

نمونه Plan:

{
  "question_interpretation": "محاسبه مجموع مبلغ سفارش‌های تکمیل‌شده به تفکیک شهر",
  "clarification_needed": false,
  "filters": [
    {
      "column": "status",
      "operator": "eq",
      "value": "completed"
    }
  ],
  "group_by": [
    "city"
  ],
  "aggregations": [
    {
      "column": "total_amount",
      "function": "sum",
      "alias": "total_sales"
    }
  ],
  "sort": [
    {
      "column": "total_sales",
      "direction": "desc"
    }
  ],
  "limit": 100,
  "assumptions": []
}

سؤال‌های مناسب برای آزمایش

تعداد سفارش‌ها در هر وضعیت چقدر است؟

متوسط مبلغ سفارش‌های تکمیل‌شده در هر شهر را نمایش بده.

پنج محصول با بیشترین تعداد فروش در سفارش‌های تکمیل‌شده کدام‌اند؟

فروش هر دسته محصول را از بیشترین به کمترین مرتب کن.

تعداد مشتری یکتا در هر شهر چقدر است؟

سفارش‌هایی با مبلغ بیشتر از ده میلیون ریال را نمایش بده.

مدیریت سؤال‌های مبهم

سؤال:

بهترین مشتری‌ها را نمایش بده.

ابهام:

• بیشترین مبلغ خرید؟
• بیشترین تعداد سفارش؟
• بیشترین تعداد محصول؟
• مشتری فعال در بازه زمانی؟
• سفارش لغوشده حذف شود؟

مدل باید سؤال تکمیلی بپرسد:

{
  "clarification_needed": true,
  "clarification_question": "بهترین مشتری را بر اساس مجموع مبلغ خرید، تعداد سفارش یا معیار دیگری تعریف می‌کنید؟"
}

تحلیل تاریخ

برای سؤال‌های ماهانه ابتدا ستون تاریخ باید به نوع datetime تبدیل شود:

def convert_date_column(
    dataframe,
    column_name: str,
):
    converted = pd.to_datetime(
        dataframe[column_name],
        errors="coerce",
        utc=False,
    )

    success_rate = converted.notna().mean()

    if success_rate < 0.95:
        raise ValueError(
            f"Column {column_name} "
            "could not be reliably parsed "
            "as a date."
        )

    dataframe = dataframe.copy()
    dataframe[column_name] = converted

    return dataframe

سپس ستون ماه به‌صورت قطعی در Backend ساخته شود:

dataframe["order_month"] = (
    dataframe["order_date"]
    .dt.to_period("M")
    .astype(str)
)

برای تاریخ شمسی باید تبدیل تقویم به‌صورت مرحله‌ای و با کتابخانه مناسب انجام شود. مدل نباید تاریخ شمسی را با حدس به میلادی تبدیل کند.

محاسبه نرخ و درصد

برای نرخ لغو:

Cancelled Orders / Total Orders × 100

بهتر است این محاسبه در کد انجام شود، نه در مدل.

total_orders = len(dataframe)

cancelled_orders = int(
    (
        dataframe["status"]
        == "cancelled"
    ).sum()
)

cancellation_rate = (
    cancelled_orders
    / total_orders
    * 100
    if total_orders
    else None
)

مدل می‌تواند نتیجه محاسبه‌شده را توضیح دهد.

تشخیص داده تکراری

تکراری کامل:

duplicate_count = int(
    dataframe.duplicated().sum()
)

تکراری بر اساس کلید:

duplicate_order_ids = (
    dataframe[
        dataframe.duplicated(
            subset=["order_id"],
            keep=False,
        )
    ]
)

دو ردیف کاملاً یکسان با دو سفارش مجزا الزاماً تکراری نیستند. تعریف Duplicate باید بر اساس کلید کسب‌وکار باشد.

کنترل Missing Value

missing_report = (
    dataframe.isna()
    .sum()
    .sort_values(
        ascending=False
    )
)

مدل نباید خودکار تصمیم بگیرد مقدار خالی:

• حذف شود.
• صفر شود.
• با میانگین جایگزین شود.
• از ردیف قبلی کپی شود.

روش برخورد با Missing Value یک تصمیم تحلیلی است و باید صریح باشد.

تحلیل Outlier

روش IQR:

def find_iqr_outliers(
    series: pd.Series,
) -> pd.Series:
    q1 = series.quantile(0.25)
    q3 = series.quantile(0.75)

    iqr = q3 - q1

    lower_bound = q1 - 1.5 * iqr
    upper_bound = q3 + 1.5 * iqr

    return (
        (series < lower_bound)
        | (series > upper_bound)
    )

Outlier الزاماً داده اشتباه نیست. ممکن است یک سفارش بزرگ و واقعی باشد. خروجی باید «مورد نیازمند بررسی» معرفی شود، نه «خطا».

ساخت نمودار

AI می‌تواند نوع نمودار پیشنهاد کند، اما داده نمودار باید از نتیجه محاسبه‌شده ساخته شود.

برای مثال:

import matplotlib.pyplot as plt


result_dataframe.plot(
    x="city",
    y="total_sales",
    kind="bar",
    legend=False,
)

plt.title("فروش تکمیل‌شده هر شهر")
plt.xlabel("شهر")
plt.ylabel("فروش به ریال")
plt.tight_layout()
plt.savefig(
    "output/sales-by-city.png",
    dpi=150,
)

نمودار باید بر اساس DataFrame نتیجه ساخته شود، نه تصویر تولیدشده توسط مدل.

ساخت گزارش مدیریتی

ورودی مدل:

{
  "question": "فروش هر شهر",
  "result": {
    "columns": [
      "city",
      "total_sales"
    ],
    "rows": [
      ["اصفهان", 52000000],
      ["شیراز", 22700000],
      ["تهران", 9500000]
    ]
  },
  "unit": "IRR"
}

خروجی:

• خلاصه نتیجه
• رتبه‌بندی
• نکات قابل مشاهده
• محدودیت داده
• سؤال تحلیلی بعدی

مدل نباید از یک جدول کوچک، علت فروش بیشتر یک شهر را حدس بزند.

چرا تولید مستقیم Python توسط مدل مناسب نیست؟

برخی ابزارها سؤال را دریافت و کد Python تولیدشده را اجرا می‌کنند. این روش انعطاف زیادی دارد، اما کنترل آن دشوارتر است.

کد ممکن است:

• فایل‌های دیگر را بخواند.
• به شبکه متصل شود.
• زمان زیادی اجرا شود.
• حافظه زیادی مصرف کند.
• داده را تغییر دهد.
• خروجی غیرقابل پیش‌بینی بسازد.
• از کتابخانه نصب‌نشده استفاده کند.

روش Plan محدود:

{
  "filters": [],
  "group_by": [],
  "aggregations": []
}

قابل اعتبارسنجی، قابل ثبت و قابل بازتولید است.

اگر اجرای Python لازم باشد، باید در Sandbox جدا، با محدودیت زمان، حافظه، فایل و شبکه انجام شود.

پردازش فایل بزرگ با DuckDB

برای CSV بزرگ می‌توان از DuckDB استفاده کرد:

pip install duckdb

نمونه:

import duckdb


result = duckdb.sql(
    """
    SELECT
        city,
        SUM(total_amount) AS total_sales
    FROM read_csv_auto('sales.csv')
    WHERE status = 'completed'
    GROUP BY city
    ORDER BY total_sales DESC
    """
).df()

در این حالت نیز Query باید از یک Plan اعتبارسنجی‌شده ساخته شود و مستقیماً SQL آزاد مدل اجرا نشود.

استفاده از Semantic Layer

برای Datasetهای سازمانی، اصطلاحات را تعریف کنید:

metrics:
  completed_sales:
    description: مجموع مبلغ سفارش‌های تکمیل‌شده
    column: total_amount
    aggregation: sum
    filters:
      - status = completed
    unit: IRR

  unique_customers:
    description: تعداد مشتری یکتا
    column: customer_id
    aggregation: nunique

dimensions:
  city:
    column: city

  product_category:
    column: category

در این حالت «فروش» به تعریف رسمی completed_sales متصل می‌شود.

حفظ تاریخچه سؤال‌ها

کاربر ممکن است بپرسد:

فروش هر شهر را نمایش بده.

سپس:

فقط سفارش‌های ماه تیر.

و بعد:

سه شهر اول را نگه دار.

به‌جای ارسال تمام گفت‌وگو، State ساختاریافته نگه دارید:

{
  "metric": "completed_sales",
  "group_by": [
    "city"
  ],
  "filters": [
    {
      "column": "order_month",
      "operator": "eq",
      "value": "2026-07"
    }
  ],
  "limit": 3
}

در سؤال بعدی مدل Plan قبلی را اصلاح می‌کند.

ارزیابی AI Data Analyst

Dataset ارزیابی باید شامل این سؤال‌ها باشد:

• Sum ساده
• Average
• Count
• Unique Count
• Group By
• چند Filter
• Sort و Limit
• مقدار خالی
• سؤال مبهم
• ستون موجودنبوده
• شناسه عددی
• تاریخ
• درصد
• نتیجه خالی
• نام ستون فارسی
• واحد پول نامشخص

معیارها:

ساخت Dataset مرجع Evals

[
  {
    "id": "q-001",
    "question": "فروش تکمیل‌شده هر شهر چقدر است؟",
    "expected_plan": {
      "filter": {
        "status": "completed"
      },
      "group_by": [
        "city"
      ],
      "aggregation": {
        "column": "total_amount",
        "function": "sum"
      }
    },
    "clarification_needed": false
  },
  {
    "id": "q-002",
    "question": "بهترین مشتری‌ها کدام‌اند؟",
    "expected_plan": null,
    "clarification_needed": true
  }
]

نتیجه نهایی را نیز با خروجی Pandas مرجع مقایسه کنید.

انتخاب مدل مناسب

مدل مناسب باید در این موارد عملکرد خوبی داشته باشد:

• درک سؤال فارسی
• پیروی از Schema
• تولید JSON معتبر
• شناخت مفاهیم تحلیل داده
• تشخیص سؤال مبهم
• انتخاب Aggregation مناسب
• حفظ نام ستون‌ها
• توضیح دقیق نتیجه

برای Planهای ساده، مدل سریع و اقتصادی مناسب است. برای سؤال‌های چندمرحله‌ای و Dataset دارای Semantic Layer پیچیده، مدل قوی‌تر ممکن است بهتر باشد.

فهرست مدل‌ها و قیمت‌های به‌روز در صفحه مدل‌های درواره قرار دارد.

اشتباهات رایج

ارسال کل CSV به مدل

Profile، Schema و نتیجه Aggregated را ارسال کنید.

سپردن محاسبات به مدل

محاسبه باید با Pandas، SQL یا کد انجام شود.

اجرای Python تولیدشده بدون محدودیت

از Analysis Plan محدود و اعتبارسنجی‌شده استفاده کنید.

حدس‌زدن واحد مبلغ

واحد را در Metadata مشخص کنید.

Sum کردن شناسه‌ها

ستون عددی الزاماً Metric نیست.

حذف خودکار Missing Value

روش برخورد با داده خالی باید صریح باشد.

معرفی Outlier به‌عنوان خطا

Outlier فقط یک مورد نیازمند بررسی است.

استفاده از Sample به‌عنوان کل Dataset

مدل باید بداند Sample فقط نمونه است و نتیجه نباید از آن محاسبه شود.

نداشتن سؤال تکمیلی

سؤال مبهم نباید به Plan حدسی تبدیل شود.

نقشه راه Production

مرحله اول: Profile

فایل بارگذاری و کیفیت داده نمایش داده شود.

مرحله دوم: سؤال‌های محدود

فقط Filter، Group By، Aggregation و Sort پشتیبانی شوند.

مرحله سوم: Semantic Layer

Metricها و Dimensionهای رسمی تعریف شوند.

مرحله چهارم: Evals

سؤال‌ها و پاسخ‌های مرجع ساخته شوند.

مرحله پنجم: نمودار

خروجی محاسبه‌شده به نمودار تبدیل شود.

مرحله ششم: Dataset بزرگ

پردازش به DuckDB یا Data Warehouse منتقل شود.

مرحله هفتم: گزارش‌های ذخیره‌شده

Plan، نتیجه، مدل و نسخه Prompt برای بازتولید ثبت شوند.

چک‌لیست تحلیل CSV با AI

• Encoding و Separator بررسی شده‌اند.
• نام ستون تکراری وجود ندارد.
• نوع داده‌ها تأیید شده است.
• واحد مبلغ مشخص است.
• تاریخ و تقویم مشخص‌اند.
• شناسه‌ها از Metricها جدا شده‌اند.
• قواعد کسب‌وکار ثبت شده‌اند.
• فقط Profile کنترل‌شده به مدل ارسال می‌شود.
• خروجی مدل JSON است.
• تمام ستون‌های Plan اعتبارسنجی می‌شوند.
• Python آزاد مدل اجرا نمی‌شود.
• محاسبات با Pandas یا SQL انجام می‌شوند.
• خروجی عددی به مدل برگردانده می‌شود.
• مدل عدد جدید اختراع نمی‌کند.
• سؤال مبهم متوقف می‌شود.
• نتیجه همراه Plan ذخیره می‌شود.
• API Key فقط در Backend قرار دارد.

پرسش‌های متداول

آیا هوش مصنوعی می‌تواند فایل CSV را تحلیل کند؟

بله. مدل می‌تواند سؤال کاربر را درک و برنامه تحلیل تولید کند. بهتر است محاسبات نهایی با Pandas، SQL یا یک موتور تحلیلی انجام شوند.

چگونه با فایل CSV چت کنیم؟

فایل را در Backend بخوانید، Schema و Profile بسازید، سؤال را به Analysis Plan تبدیل و نتیجه محاسبه‌شده را به زبان طبیعی توضیح دهید.

آیا می‌توان CSV فارسی را تحلیل کرد؟

بله. باید Encoding، نام ستون‌های فارسی، ارقام فارسی و نوع تاریخ بررسی و نرمال شوند.

چرا نباید فایل کامل را برای مدل بفرستیم؟

فایل ممکن است بزرگ باشد، هزینه افزایش می‌یابد و مدل برای محاسبات دقیق روی هزاران ردیف مناسب نیست.

آیا AI می‌تواند نمودار بسازد؟

مدل می‌تواند نوع نمودار را پیشنهاد کند. داده و تصویر نمودار بهتر است با Pandas، Matplotlib یا ابزار مشابه ساخته شوند.

آیا می‌توان چند فایل CSV را با هم تحلیل کرد؟

بله. ابتدا باید کلید Join، نوع رابطه و کیفیت تطبیق مشخص شود. مدل نباید ستون Join را بدون شواهد حدس بزند.

برای فایل بزرگ چه کنیم؟

از Chunk Processing، DuckDB، دیتابیس تحلیلی یا Data Warehouse استفاده کنید و فقط نتیجه Aggregated را به مدل بدهید.

بهترین مدل برای تحلیل CSV چیست؟

مدل باید در درک فارسی، تحلیل داده و تولید Structured Output مناسب باشد. مدل‌های موجود را در صفحه مدل‌های درواره مقایسه کنید.

API درواره چگونه متصل می‌شود؟

در Backend، base_url را برابر https://api.darvareh.ir/v1 قرار دهید و Dataset Profile و سؤال را برای تولید Analysis Plan به مدل بفرستید.

جمع‌بندی

تحلیل CSV با هوش مصنوعی زمانی قابل اعتماد است که وظیفه مدل و موتور محاسبات از یکدیگر جدا باشند.

مدل هوش مصنوعی باید سؤال فارسی را بفهمد، ابهام را تشخیص دهد و یک Analysis Plan محدود بسازد. Pandas، SQL یا DuckDB باید فیلتر، Group By، Sum، Average، Sort و سایر محاسبات را اجرا کنند. در پایان مدل می‌تواند نتیجه قطعی را به زبان ساده توضیح دهد.

در پروژه این مقاله یک AI Data Analyst با Python، Pandas، FastAPI، Pydantic و API درواره ساختیم. این سیستم CSV را پروفایل می‌کند، فقط خلاصه کنترل‌شده را به مدل می‌فرستد، Plan را اعتبارسنجی و سپس آن را بدون اجرای کد آزاد مدل اجرا می‌کند.

برای شروع، در درواره ثبت‌نام و API Key دریافت کنید. سپس مدل مناسب را از صفحه مدل‌های درواره انتخاب و پروژه را ابتدا با یک CSV کوچک و دارای تعریف روشن ستون‌ها آزمایش کنید.

مقالات مرتبط

• هوش مصنوعی برای اکسل؛ ساخت فرمول و تحلیل داده
• آموزش Structured Outputs و JSON Schema
• آموزش ارزیابی مدل‌های هوش مصنوعی و Evals
• چگونه API هوش مصنوعی را به نرم‌افزار خود اضافه کنیم؟
• ساخت سیستم تحلیل بازخورد مشتری با هوش مصنوعی
• راهنمای ساخت API هوش مصنوعی آماده Production
• راهنمای کاهش هزینه API هوش مصنوعی
• راهنمای انتخاب بهترین API هوش مصنوعی

برای مطالعه شرایط استفاده و محدودیت‌های مسئولیت، صفحه «سلب مسئولیت» را مشاهده کنید.