シニアアナリスト並みの思考を持つ AI データ分析ツールの構築方法

import pandas as pd
from IPython.display import display
orders = pd.read_csv("online_orders.csv")
print(f"Loaded {len(orders):,} rows and {len(orders.columns)} columns.")
display(orders.head())

Loaded 29 rows and 6 columns.

schema_preview = pd.DataFrame({
    "column": orders.columns,
    "dtype": orders.dtypes.astype(str).values,
    "missing_values": orders.isna().sum().values,
})
display(schema_preview)

import duckdb
con = duckdb.connect()
con.register("online_orders", orders)
preview = con.execute("""
    SELECT
        promotion_id,
        COUNT(*) AS n_orders,
        SUM(units_sold) AS total_units,
        SUM(cost_in_dollars * units_sold) AS total_revenue,
        ROUND(AVG(units_sold), 2) AS avg_units_per_order
    FROM online_orders
    GROUP BY promotion_id
    ORDER BY avg_units_per_order DESC
""").df()
display(preview)

class LLMClient:
    def __init__(self, client, model, provider):
        self.client = client
        self.model = model
        self.provider = provider

    def complete(self, prompt):
        if self.provider == "anthropic":
            response = self.client.messages.create(
                model=self.model,
                max_tokens=1024,
                messages=[{"role": "user", "content": prompt}],
            )

            for block in response.content:
                if block.type == "text":
                    return block.text

            raise ValueError("No text block found in Claude's response.")

        if self.provider == "openai":
            response = self.client.chat.completions.create(
                model=self.model,
                messages=[{"role": "user", "content": prompt}],
            )
            return response.choices[0].message.content

        raise ValueError(f"Unsupported provider: {self.provider}")

import json
import re
def parse_json(text):
    text = text.strip()

    if text.startswith("```

"): text = re.sub(r"^ ```(?:json)?\s*", "", text, flags=re.IGNORECASE)
        text = re.sub(r"\s*```$", "", text)

try:
        return json.loads(text)
    except json.JSONDecodeError:
        pass
    candidates = []
    object_match = re.search(r"\{.*\}", text, re.DOTALL)
    array_match = re.search(r"\[.*\]", text, re.DOTALL)
    if object_match:
        candidates.append(object_match)
    if array_match:
        candidates.append(array_match)
    candidates.sort(key=lambda match: match.start())
    for match in candidates:
        try:
            return json.loads(match.group(0))
        except json.JSONDecodeError:
            continue
    raise ValueError(f"No valid JSON found in model output:\n{text}")

class SeniorAnalyst:
    MIN_SUPPORT = 3  # minimum orders behind a group before we trust it

def __init__(self, llm, table_name, dataframe):
        self.llm = llm
        self.table_name = table_name
        self.con = duckdb.connect()
        self.con.register(table_name, dataframe)
        self.schema = self.con.execute(f"DESCRIBE {table_name}").df()

def understand_business_context(self, question):
        row_count = self.con.execute(
            f"SELECT COUNT(*) FROM {self.table_name}"
        ).fetchone()[0]

columns = self.schema[
            ["column_name", "column_type"]
        ].to_dict("records")
        prompt = f"""You are a senior data analyst. A stakeholder asked: "{question}"
Table: {self.table_name}
Columns: {columns}
Row count: {row_count}

Restate the stakeholder question in terms this table can actually answer.

また、テーブルの粒度（1 行が何を表すか）を明記し、すでに把握できる制約事項も列挙してください。具体的には、サンプルサイズ、日付のカバレッジ範囲、欠落している次元や文脈などです。

リファインされた質問、粒度、制限事項を返す JSON 形式の出力が必要です。

```python
context = parse_json(self.llm.complete(prompt))
self.context = context
return context

We ran this with claude-sonnet-5 on the question "which promotion should we run more of." Here is what came back.

## Output

![Build AI Data Analyst](https://www.kdnuggets.com/wp-content/uploads/Rosidi-AI-Data-Analyst-Senior-4.png)

It flagged the small sample size before running a single query — the same trap the plain SQL `groupby` above already showed us. That flag is a hint, not a check. The pipeline still needs to enforce it in code, which is what the validation stage below does.

## Stage 2: Hypothesis Generation

The second stage proposes specific, testable hypotheses using only the columns that exist in the table.

def generate_hypotheses(self, n=2):
    columns = list(self.schema["column_name"])
    prompt = f"""Business context: {self.context}
Propose {n} specific, testable hypotheses that would help answer the
restated question, using only columns in: {columns}.
Each hypothesis should be something we can test using SQL.
Return JSON only: [{{"hypothesis": "...", "why": "..."}}, ...]"""
    hypotheses = parse_json(self.llm.complete(prompt))
    self.hypotheses = hypotheses
    return hypotheses

## Output

![Build AI Data Analyst](https://www.kdnuggets.com/wp-content/uploads/Rosidi-AI-Data-Analyst-Senior-5.png)

The pipeline tests the first hypothesis. Notice it is not a raw average: it asks whether the volume leader beats the runner-up by a real margin, which already reads differently from the "highest average" query above that put a 1-order promotion on top.

## Stage 3: SQL Planning

The third stage turns the top hypothesis into an actual query. We ask for a row count alongside any grouped metric, since a group's size is what the validation stage checks next.

def plan_sql(self, hypothesis):
    columns = list(self.schema["column_name"])
    prompt = f"""Table: {self.table_name}
Columns: {columns}
Hypothesis to test: {hypothesis['hypothesis']}
Write one DuckDB SQL query that tests this hypothesis.
Use only the available columns, do not invent columns, and if the query
groups rows, include a COUNT(*) column named n_orders so the result can
be checked for sample size before anyone trusts it.
Return JSON only: {{"sql": "...", "purpose": "..."}}"""
    plan = parse_json(self.llm.complete(prompt))
    return plan

## Output

Generated SQL:
    WITH promo_sums AS (
        SELECT promotion_id, SUM(units_sold) AS total_units, COUNT(*) AS n_orders
        FROM online_orders
        GROUP BY promotion_id
    ),
    ranked AS (

Rather than a simple `groupby`, the model reached for a common table expression (CTE) with a window function, ranking promotions by total units and pulling the top two into the same row for comparison.

## Stage 4: Validation

The fourth stage runs the query and checks `n_orders` against a minimum support threshold. This is the one stage that is plain code, not a model call, because the check has to be enforced, not suggested.

## Output

| top_promotion_id | top_total_units | top_n_orders | second_promotion_id | second_total_units | second_n_orders | pct_difference | low_confidence |
| --- | --- | --- | --- | --- | --- | --- | --- |
| 1 | 77.0 | 12 | 2 | 55.0 | 10 | 0.4 | False |

This query only produces one row, and it is not flagged. Promotion 1 leads on total units with 12 orders behind it, promotion 2 is the runner-up with 10, and both clear the minimum of 3 we set. The check still ran here — it just had nothing to catch, because this hypothesis compares two well-supported groups instead of resting on promotion 4's single order.

## Stage 5: Executive Summary

The fifth stage writes the summary, and it is told explicitly to leave any flagged row out of the headline claim.

## Output

![Build AI Data Analyst](https://www.kdnuggets.com/wp-content/uploads/Rosidi-AI-Data-Analyst-Senior-6.png)

## Stage 6: Recommendations

The sixth stage proposes actions, and it is told the same rule applies: no recommendation may rest on low-confidence data or facts the summary did not support.

## Output

![Build AI Data Analyst](https://www.kdnuggets.com/wp-content/uploads/Rosidi-AI-Data-Analyst-Senior-7.png)

## Putting It Together

A `run` method chains the six stages. One call takes a question in and returns every intermediate result: the context, the hypotheses, the SQL plan, the validated table, the summary, and the recommendation.

![Build AI Data Analyst](https://www.kdnuggets.com/wp-content/uploads/Rosidi-AI-Data-Analyst-Senior-8.png)

## Calling It

Calling it looks the same regardless of which provider you bring. The provider is set explicitly rather than guessed from the client object, and the pipeline refuses to run if you forget to paste in a real key.

プロバイダーを `openai` に設定し、OpenAI のキーを追加すれば、同じ 6 つのステージが `gpt-4o` で変更なく実行されます。 （原文の技術表記: `PROVIDER`）

`LLMClient` は、自分がどの API と通信しているかを知る唯一のコンポーネントです。

## 結論

ここで紹介する 6 つの工程は、それぞれ単独で見れば決して複雑ではありません。質問を言い換える、SQL を記述する、表を要約するといった作業は、すでに一つのプロンプトでも十分にこなせるレベルです。

真に価値を生むのは、クエリと要約の間に挟まれる検証工程です。例えば、`n_orders` の値を確認してから、初めてそれを「答え」として扱うという手順が重要です。

このデータセットでは、LLM が呼び出される前ですでにチェックが機能していました。上記の単純な SQL の `groupby` 処理では、1 件しかない注文に基づいて平均注文単価が最も高い「プロモーション 4」が 1 位にランクインしています。しかし、モデルがこの実行をテストするために選択した仮説は、より根拠のある 2 つのグループ（12 件対 10 件の注文）を比較するものであり、`validate()` 関数には警告を出す要素がありませんでした。

このパイプラインでは、モデルが提示する比較内容に関わらず、常に同じ `n_orders` チェックを実行します。そのため、将来のテーブルや、合計値ではなく平均値を検証する実行においても、同様のコード行によって問題が発見されることになります。

このパイプラインには 1 つのクラスに 6 つのメソッドが定義されており、対象とする次のテーブルに対しても同じ 6 つのメソッドが再度実行されます。