Gemini 3 Pro × LangGraph で構築する自律型データ分析エージェント

取り組みの背景

データ分析の現場では、「CSV を読み込んで統計を出し、グラフを作成して、最後にレポートをまとめる」という一連の作業を何度も繰り返すことがあります。こうしたマルチステップの分析ワークフローを AI に任せられたら、どれほど生産性が向上するでしょうか。

この記事で学べること:

LangGraph のステートグラフを使ったエージェント設計パターン
Gemini 3 Pro API と LangGraph の統合方法
CSV 解析 → 統計分析 → 可視化 → レポート生成の自動パイプライン
エラーハンドリングとリトライ戦略
プロダクション環境へのデプロイのベストプラクティス

対象読者: Python と Gemini API の基本を理解している中〜上級者

LangGraph とは何か

LangGraph は、LangChain チームが開発したステートフルなマルチアクターアプリケーションを構築するためのフレームワークです。従来の LangChain の Chain（直線的なパイプライン）とは異なり、LangGraph では有向グラフとしてワークフローを定義します。

LangGraph が AI エージェント構築に適している理由は次の通りです。

ステート管理: 各ステップ間で状態を保持し、前のステップの結果を次のステップで参照できる
条件分岐: 分析結果に応じて次のアクションを動的に変更できる
ループ処理: 結果が不十分な場合に自動でリトライや再分析を実行できる
人間介入ポイント: 重要な判断で人間の承認を挟むことが可能

環境準備

必要なパッケージのインストール

pip install google-genai langgraph langchain-google-genai pandas matplotlib seaborn

API キーの設定

Google AI Studio で API キーを取得し、環境変数に設定します。

import os
os.environ["GOOGLE_API_KEY"] = "your-api-key-here"

動作確認

まず Gemini 3 Pro API が正しく動作するか確認しましょう。

from google import genai
 
client = genai.Client()
response = client.models.generate_content(
    model="gemini-3-pro",
    contents="Pythonでデータ分析を行う際のベストプラクティスを3つ教えてください。"
)
print(response.text)

期待する出力:

1. データの前処理を徹底する — 欠損値や外れ値の処理を最初に行うことで...
2. 探索的データ分析（EDA）を怠らない — 可視化を通じてデータの全体像を...
3. 再現性を確保する — ランダムシードの固定やバージョン管理を...

エージェントのアーキテクチャ設計

今回構築するデータ分析エージェントは、以下の 5 つのノードで構成されるステートグラフです。

データ読み込みノード — CSV ファイルを読み込み、基本情報を取得
分析計画ノード — Gemini 3 Pro がデータを分析して実行計画を立案
統計分析ノード — 計画に基づいて統計処理を実行
可視化ノード — 分析結果をグラフとして描画
レポート生成ノード — Gemini 3 Pro が最終レポートを自然言語で作成

ステート定義

from typing import TypedDict, Annotated, Sequence
from langgraph.graph import StateGraph, END
import pandas as pd
 
class AnalysisState(TypedDict):
    """エージェント全体で共有される状態"""
    file_path: str                    # 入力CSVファイルパス
    dataframe: pd.DataFrame | None    # 読み込んだデータ
    data_summary: str                 # データの概要
    analysis_plan: list[str]          # 分析計画のステップリスト
    statistics: dict                  # 統計分析の結果
    chart_paths: list[str]           # 生成したグラフのファイルパス
    report: str                       # 最終レポート
    error: str | None                # エラー情報
    retry_count: int                  # リトライ回数

各ノードの実装

ノード 1: データ読み込み

import pandas as pd
import json
 
def load_data_node(state: AnalysisState) -> AnalysisState:
    """CSVファイルを読み込み、基本的なデータサマリーを生成する"""
    try:
        df = pd.read_csv(state["file_path"])
 
        # データの概要を文字列で生成
        summary_parts = [
            f"行数: {len(df)}, 列数: {len(df.columns)}",
            f"カラム: {', '.join(df.columns.tolist())}",
            f"データ型:\n{df.dtypes.to_string()}",
            f"欠損値:\n{df.isnull().sum().to_string()}",
            f"先頭5行:\n{df.head().to_string()}",
            f"基本統計量:\n{df.describe().to_string()}"
        ]
        data_summary = "\n\n".join(summary_parts)
 
        return {
            **state,
            "dataframe": df,
            "data_summary": data_summary,
            "error": None
        }
    except Exception as e:
        return {**state, "error": f"データ読み込みエラー: {str(e)}"}

ノード 2: 分析計画の立案（Gemini 3 Pro）

このノードでは、Gemini 3 Pro にデータの概要を渡し、最適な分析計画を立案してもらいます。

from google import genai
 
client = genai.Client()
 
def plan_analysis_node(state: AnalysisState) -> AnalysisState:
    """Gemini 3 Pro がデータを分析し、実行計画を立案する"""
    if state.get("error"):
        return state
 
    prompt = f"""あなたはデータ分析の専門家です。
以下のデータセットの概要を確認し、有意義な分析計画を JSON 形式で作成してください。
 
## データ概要
{state['data_summary']}
 
## 出力形式
以下の JSON 形式で回答してください:
{{
  "analysis_steps": [
    {{
      "step": "ステップの説明",
      "method": "使用する統計手法",
      "columns": ["対象カラム"]
    }}
  ],
  "recommended_charts": [
    {{
      "type": "グラフの種類（histogram, scatter, heatmap, bar, line）",
      "columns": ["対象カラム"],
      "title": "グラフのタイトル"
    }}
  ]
}}
"""
 
    response = client.models.generate_content(
        model="gemini-3-pro",
        contents=prompt,
        config={
            "response_mime_type": "application/json"
        }
    )
 
    try:
        plan = json.loads(response.text)
        analysis_steps = [step["step"] for step in plan["analysis_steps"]]
        return {
            **state,
            "analysis_plan": analysis_steps,
            "_full_plan": plan  # 内部用に完全な計画を保持
        }
    except json.JSONDecodeError:
        return {**state, "error": "分析計画の解析に失敗しました"}

ノード 3: 統計分析の実行

import numpy as np
 
def run_statistics_node(state: AnalysisState) -> AnalysisState:
    """分析計画に基づいて統計処理を実行する"""
    if state.get("error"):
        return state
 
    df = state["dataframe"]
    stats = {}
 
    # 数値カラムの自動検出
    numeric_cols = df.select_dtypes(include=[np.number]).columns.tolist()
    categorical_cols = df.select_dtypes(include=["object", "category"]).columns.tolist()
 
    # 基本統計量
    stats["basic"] = {
        col: {
            "mean": float(df[col].mean()),
            "median": float(df[col].median()),
            "std": float(df[col].std()),
            "min": float(df[col].min()),
            "max": float(df[col].max()),
            "skew": float(df[col].skew()),
            "kurtosis": float(df[col].kurtosis())
        }
        for col in numeric_cols
    }
 
    # 相関分析（数値カラムが2つ以上ある場合）
    if len(numeric_cols) >= 2:
        corr_matrix = df[numeric_cols].corr()
        stats["correlation"] = corr_matrix.to_dict()
 
        # 強い相関を検出（|r| > 0.7）
        strong_correlations = []
        for i in range(len(numeric_cols)):
            for j in range(i + 1, len(numeric_cols)):
                r = corr_matrix.iloc[i, j]
                if abs(r) > 0.7:
                    strong_correlations.append({
                        "col1": numeric_cols[i],
                        "col2": numeric_cols[j],
                        "correlation": round(r, 4)
                    })
        stats["strong_correlations"] = strong_correlations
 
    # カテゴリカルデータの集計
    if categorical_cols:
        stats["categorical"] = {
            col: df[col].value_counts().head(10).to_dict()
            for col in categorical_cols
        }
 
    return {**state, "statistics": stats}

ノード 4: 可視化

import matplotlib.pyplot as plt
import matplotlib
import seaborn as sns
import os
 
# 日本語フォント設定
matplotlib.rcParams['font.family'] = 'DejaVu Sans'
 
def create_visualizations_node(state: AnalysisState) -> AnalysisState:
    """統計結果に基づいてグラフを生成する"""
    if state.get("error"):
        return state
 
    df = state["dataframe"]
    chart_paths = []
    output_dir = "analysis_output"
    os.makedirs(output_dir, exist_ok=True)
 
    numeric_cols = df.select_dtypes(include=[np.number]).columns.tolist()
 
    # 1. ヒストグラム（数値カラムの分布）
    if numeric_cols:
        fig, axes = plt.subplots(
            1, min(len(numeric_cols), 4),
            figsize=(5 * min(len(numeric_cols), 4), 4)
        )
        if len(numeric_cols) == 1:
            axes = [axes]
        for idx, col in enumerate(numeric_cols[:4]):
            axes[idx].hist(df[col].dropna(), bins=30, edgecolor='black', alpha=0.7)
            axes[idx].set_title(f'Distribution: {col}')
            axes[idx].set_xlabel(col)
            axes[idx].set_ylabel('Frequency')
        plt.tight_layout()
        path = f"{output_dir}/histograms.png"
        plt.savefig(path, dpi=150, bbox_inches='tight')
        plt.close()
        chart_paths.append(path)
 
    # 2. 相関ヒートマップ
    if len(numeric_cols) >= 2:
        fig, ax = plt.subplots(figsize=(10, 8))
        corr = df[numeric_cols].corr()
        sns.heatmap(corr, annot=True, fmt=".2f", cmap="coolwarm",
                    center=0, ax=ax, square=True)
        ax.set_title("Correlation Heatmap")
        plt.tight_layout()
        path = f"{output_dir}/correlation_heatmap.png"
        plt.savefig(path, dpi=150, bbox_inches='tight')
        plt.close()
        chart_paths.append(path)
 
    # 3. 散布図（強い相関がある組み合わせ）
    strong_corrs = state["statistics"].get("strong_correlations", [])
    for i, corr_info in enumerate(strong_corrs[:3]):
        fig, ax = plt.subplots(figsize=(8, 6))
        ax.scatter(
            df[corr_info["col1"]],
            df[corr_info["col2"]],
            alpha=0.5, edgecolors='black', linewidth=0.5
        )
        ax.set_xlabel(corr_info["col1"])
        ax.set_ylabel(corr_info["col2"])
        ax.set_title(
            f'{corr_info["col1"]} vs {corr_info["col2"]} '
            f'(r={corr_info["correlation"]})'
        )
        plt.tight_layout()
        path = f"{output_dir}/scatter_{i+1}.png"
        plt.savefig(path, dpi=150, bbox_inches='tight')
        plt.close()
        chart_paths.append(path)
 
    return {**state, "chart_paths": chart_paths}

ノード 5: レポート生成（Gemini 3 Pro）

def generate_report_node(state: AnalysisState) -> AnalysisState:
    """Gemini 3 Pro が統計結果を基にレポートを生成する"""
    if state.get("error"):
        return state
 
    stats_summary = json.dumps(state["statistics"], ensure_ascii=False, indent=2)
 
    prompt = f"""あなたはデータアナリストです。以下の統計分析結果を基に、
ビジネスパーソンが理解できる形式でレポートを作成してください。
 
## データ概要
{state['data_summary'][:2000]}
 
## 統計分析結果
{stats_summary[:3000]}
 
## 生成されたグラフ
{', '.join(state['chart_paths'])}
 
## レポート要件
1. エグゼクティブサマリー（3行以内）
2. 主要な発見事項（箇条書き）
3. データの傾向と特徴
4. 相関関係の解釈（ビジネス的な意味を含める）
5. 推奨アクション
6. 注意事項・データの限界
 
専門用語は避け、具体的な数値を引用してください。
"""
 
    response = client.models.generate_content(
        model="gemini-3-pro",
        contents=prompt
    )
 
    return {**state, "report": response.text}

LangGraph でワークフローを組み立てる

各ノードを LangGraph のステートグラフとして接続します。

from langgraph.graph import StateGraph, END
 
def should_retry(state: AnalysisState) -> str:
    """エラー発生時にリトライするか判定する"""
    if state.get("error") and state.get("retry_count", 0) < 3:
        return "retry"
    elif state.get("error"):
        return "fail"
    return "continue"
 
# グラフの構築
workflow = StateGraph(AnalysisState)
 
# ノードの追加
workflow.add_node("load_data", load_data_node)
workflow.add_node("plan_analysis", plan_analysis_node)
workflow.add_node("run_statistics", run_statistics_node)
workflow.add_node("create_visualizations", create_visualizations_node)
workflow.add_node("generate_report", generate_report_node)
 
# エッジの定義
workflow.set_entry_point("load_data")
 
workflow.add_conditional_edges(
    "load_data",
    should_retry,
    {
        "continue": "plan_analysis",
        "retry": "load_data",
        "fail": END
    }
)
 
workflow.add_conditional_edges(
    "plan_analysis",
    should_retry,
    {
        "continue": "run_statistics",
        "retry": "plan_analysis",
        "fail": END
    }
)
 
workflow.add_edge("run_statistics", "create_visualizations")
workflow.add_edge("create_visualizations", "generate_report")
workflow.add_edge("generate_report", END)
 
# コンパイル
app = workflow.compile()

エージェントの実行

# 初期状態を設定して実行
initial_state = {
    "file_path": "sales_data.csv",
    "dataframe": None,
    "data_summary": "",
    "analysis_plan": [],
    "statistics": {},
    "chart_paths": [],
    "report": "",
    "error": None,
    "retry_count": 0
}
 
# エージェントを実行
result = app.invoke(initial_state)
 
# レポートを出力
if result.get("error"):
    print(f"エラーが発生しました: {result['error']}")
else:
    print("=" * 60)
    print("データ分析レポート")
    print("=" * 60)
    print(result["report"])
    print(f"\n生成されたグラフ: {result['chart_paths']}")