Gemini API × Python で AI データ分析エージェントを構築する — Code Execution・Function Calling・Structured Output 統合ガイド

取り組みの背景

データ分析は多くのビジネスで不可欠なプロセスですが、従来は「データの前処理 → 探索的分析 → 可視化 → レポート作成」という一連のワークフローを手動で行う必要がありましました。Gemini API の Code Execution、Function Calling、Structured Output を組み合わせることで、この一連のプロセスを自然言語の指示だけで自動実行する AI データ分析エージェントを構築できます。

ここではCSV や Excel ファイルをアップロードするだけで、データの概要把握・統計分析・グラフ生成・インサイトの構造化出力までを自動で行うエージェントを Python でゼロから実装します。

この記事で学べること:

Gemini API の Code Execution 機能でサンドボックス内の Python コードを実行する方法
Function Calling でエージェントに外部ツール（ファイル読み込み、DB クエリ等）を接続する設計パターン
Structured Output で分析結果を型安全な JSON として取得するテクニック
これら 3 つの機能を統合した本格的なデータ分析パイプラインの構築手順

対象読者: Gemini API の基本操作を理解しており、Python でのアプリケーション開発経験がある中〜上級者。

前提知識・環境構築

必要な環境

Python 3.10 以上
Google AI Python SDK（google-genai v1.55.0 以上）
Gemini API キー（Google AI Studio で取得可能）

セットアップ

pip install google-genai pandas openpyxl matplotlib

API キーを環境変数に設定します:

export GOOGLE_API_KEY="your-api-key-here"

Python コードで SDK を初期化:

from google import genai
 
client = genai.Client()
 
# 使用するモデル（Code Execution 対応）
MODEL_ID = "gemini-3-flash"

モデル選択のポイント: Code Execution は gemini-3-flash と gemini-3-pro で利用可能です。コスト効率を重視する場合は Flash、複雑な推論が必要な場合は Pro を選択してください。

概念と設計思想

3 つの機能の役割分担

今回構築するエージェントでは、Gemini API の 3 つの機能をそれぞれ明確な役割で使い分けます。

機能	役割	具体例
Code Execution	データ処理・統計計算・グラフ生成	Pandas での集計、Matplotlib での可視化
Function Calling	外部リソースへのアクセス	ファイル読み込み、DB クエリ、API 呼び出し
Structured Output	分析結果の構造化	JSON スキーマに沿ったインサイト出力

アーキテクチャ概要

ユーザー（自然言語の質問）
    ↓
Gemini API エージェント
    ├── Function Calling → ファイル/DB からデータ取得
    ├── Code Execution  → Pandas/Matplotlib で分析・可視化
    └── Structured Output → 結果を JSON で構造化
    ↓
構造化されたレポート + グラフ画像

この設計により、ユーザーは「先月の売上データを分析して、トレンドとトップ 5 カテゴリをレポートして」と指示するだけで、エージェントが自律的にデータ取得 → 分析 → 可視化 → レポート生成を完了します。

ステップバイステップ実装

Step 1: Code Execution でデータ分析を実行する

まず、Code Execution の基本的な使い方を確認します。Gemini にデータと分析指示を渡し、サンドボックス内で Python コードを実行させます。

from google import genai
from google.genai import types
 
client = genai.Client()
 
# サンプルCSVデータ（実際にはファイルアップロードで渡す）
csv_data = """date,product,revenue,quantity
2026-01-01,Widget A,15000,120
2026-01-02,Widget A,18000,145
2026-01-03,Widget B,22000,90
2026-01-04,Widget A,16500,130
2026-01-05,Widget B,25000,105
2026-01-06,Widget C,8000,200
2026-01-07,Widget A,19000,155
2026-01-08,Widget B,21000,88
2026-01-09,Widget C,9500,220
2026-01-10,Widget A,17500,140"""
 
response = client.models.generate_content(
    model="gemini-3-flash",
    contents=f"""以下のCSVデータを分析してください。
1. 各商品の売上合計と平均単価を計算
2. 日別の売上トレンドをMatplotlibでグラフ化
3. 最も売上が高い商品とその理由を考察
 
データ:
{csv_data}""",
    config=types.GenerateContentConfig(
        tools=[types.Tool(code_execution=types.ToolCodeExecution)]
    )
)
 
# レスポンスにはテキスト、コード、実行結果、画像が含まれる
for part in response.candidates[0].content.parts:
    if part.text:
        print("📝 テキスト:", part.text)
    elif part.executable_code:
        print("💻 コード:", part.executable_code.code)
    elif part.code_execution_result:
        print("📊 結果:", part.code_execution_result.output)
    elif part.inline_data:
        # Matplotlibのグラフ画像（base64）
        print("🖼️ グラフ画像を取得しました")

期待される出力:

💻 コード: import pandas as pd
import matplotlib.pyplot as plt
...
📊 結果:
商品別売上合計:
  Widget A: ¥86,000（平均単価: ¥124）
  Widget B: ¥68,000（平均単価: ¥240）
  Widget C: ¥17,500（平均単価: ¥42）
🖼️ グラフ画像を取得しました
📝 テキスト: Widget A が最も売上が高く...

ポイント: Code Execution のサンドボックスには Pandas、NumPy、Matplotlib、scikit-learn 等が事前インストールされています。外部ライブラリの追加インストールはできません。

Step 2: Function Calling でデータソースを接続する

次に、Function Calling を使ってエージェントが外部データソースにアクセスできるようにします。

from google.genai import types
import pandas as pd
import json
 
# --- ツール定義 ---
 
def load_csv_file(file_path: str) -> str:
    """CSVファイルを読み込み、先頭20行をプレビューとして返す"""
    df = pd.read_csv(file_path)
    summary = {
        "rows": len(df),
        "columns": list(df.columns),
        "dtypes": {col: str(dtype) for col, dtype in df.dtypes.items()},
        "preview": df.head(20).to_csv(index=False),
        "describe": df.describe().to_json()
    }
    return json.dumps(summary, ensure_ascii=False)
 
def query_database(sql: str) -> str:
    """SQLクエリを実行し結果をCSV形式で返す（例: SQLite）"""
    import sqlite3
    conn = sqlite3.connect("sales.db")
    df = pd.read_sql_query(sql, conn)
    conn.close()
    return df.to_csv(index=False)
 
def save_report(filename: str, content: str) -> str:
    """分析レポートをMarkdownファイルとして保存する"""
    with open(filename, "w", encoding="utf-8") as f:
        f.write(content)
    return f"レポートを {filename} に保存しました"
 
# --- Function宣言 ---
 
tools = types.Tool(
    function_declarations=[
        types.FunctionDeclaration(
            name="load_csv_file",
            description="CSVファイルを読み込み、統計情報とプレビューを返す",
            parameters=types.Schema(
                type="OBJECT",
                properties={
                    "file_path": types.Schema(
                        type="STRING",
                        description="読み込むCSVファイルのパス"
                    )
                },
                required=["file_path"]
            )
        ),
        types.FunctionDeclaration(
            name="query_database",
            description="SQLiteデータベースにSQLクエリを実行する",
            parameters=types.Schema(
                type="OBJECT",
                properties={
                    "sql": types.Schema(
                        type="STRING",
                        description="実行するSQLクエリ"
                    )
                },
                required=["sql"]
            )
        ),
        types.FunctionDeclaration(
            name="save_report",
            description="分析レポートをファイルに保存する",
            parameters=types.Schema(
                type="OBJECT",
                properties={
                    "filename": types.Schema(
                        type="STRING",
                        description="保存先のファイル名"
                    ),
                    "content": types.Schema(
                        type="STRING",
                        description="レポートの内容（Markdown形式）"
                    )
                },
                required=["filename", "content"]
            )
        )
    ]
)

Step 3: Structured Output で分析結果を型安全に取得する

分析結果を構造化された JSON として取得する仕組みを追加します。

from google.genai import types
 
# 分析レポートのスキーマ定義
analysis_schema = types.Schema(
    type="OBJECT",
    properties={
        "summary": types.Schema(
            type="STRING",
            description="分析結果の概要（3文以内）"
        ),
        "key_metrics": types.Schema(
            type="ARRAY",
            items=types.Schema(
                type="OBJECT",
                properties={
                    "name": types.Schema(type="STRING"),
                    "value": types.Schema(type="STRING"),
                    "trend": types.Schema(
                        type="STRING",
                        enum=["up", "down", "stable"]
                    )
                }
            ),
            description="主要指標のリスト"
        ),
        "insights": types.Schema(
            type="ARRAY",
            items=types.Schema(
                type="OBJECT",
                properties={
                    "title": types.Schema(type="STRING"),
                    "description": types.Schema(type="STRING"),
                    "confidence": types.Schema(
                        type="STRING",
                        enum=["high", "medium", "low"]
                    ),
                    "action_item": types.Schema(type="STRING")
                }
            ),
            description="発見されたインサイト"
        ),
        "anomalies": types.Schema(
            type="ARRAY",
            items=types.Schema(
                type="STRING"
            ),
            description="検出された異常値やアノマリー"
        )
    },
    required=["summary", "key_metrics", "insights"]
)
 
# Structured Outputで分析結果を取得
response = client.models.generate_content(
    model="gemini-3-flash",
    contents="先ほどの売上データの分析結果を構造化して出力してください",
    config=types.GenerateContentConfig(
        response_mime_type="application/json",
        response_schema=analysis_schema
    )
)
 
import json
result = json.loads(response.text)
print(json.dumps(result, indent=2, ensure_ascii=False))

期待される出力:

{
  "summary": "Widget Aが売上トップで全体の49%を占める。Widget Bは高単価戦略で安定。Widget Cは低価格・高数量モデル。",
  "key_metrics": [
    {"name": "総売上", "value": "¥171,500", "trend": "up"},
    {"name": "日平均売上", "value": "¥17,150", "trend": "stable"},
    {"name": "最高売上日", "value": "2026-01-05 (¥25,000)", "trend": "up"}
  ],
  "insights": [
    {
      "title": "Widget Bの高単価戦略が有効",
      "description": "Widget Bは数量が最も少ないが、単価が最も高く利益率が高い可能性がある",
      "confidence": "high",
      "action_item": "Widget Bの利益率を詳細に分析し、価格戦略を最適化する"
    }
  ],
  "anomalies": ["Widget Cの1/9の売上急増（前日比+18.75%）"]
}

Step 4: 3 つの機能を統合したエージェントクラス

これまでの要素を統合し、完全なデータ分析エージェントを構築します。

from google import genai
from google.genai import types
import pandas as pd
import json
import base64
from pathlib import Path
 
class DataAnalysisAgent:
    """Gemini API を活用した AI データ分析エージェント"""
 
    def __init__(self, model: str = "gemini-3-flash"):
        self.client = genai.Client()
        self.model = model
        self.chat = None
        self._setup_tools()
 
    def _setup_tools(self):
        """Function Calling + Code Execution のツールを設定"""
        self.function_tool = types.Tool(
            function_declarations=[
                types.FunctionDeclaration(
                    name="load_csv_file",
                    description="CSVファイルを読み込み統計情報を返す",
                    parameters=types.Schema(
                        type="OBJECT",
                        properties={
                            "file_path": types.Schema(
                                type="STRING",
                                description="CSVファイルのパス"
                            )
                        },
                        required=["file_path"]
                    )
                )
            ]
        )
        self.code_exec_tool = types.Tool(
            code_execution=types.ToolCodeExecution
        )
 
    def _handle_function_call(self, fn_call) -> str:
        """Function Call のディスパッチ"""
        name = fn_call.name
        args = dict(fn_call.args)
 
        if name == "load_csv_file":
            df = pd.read_csv(args["file_path"])
            return json.dumps({
                "rows": len(df),
                "columns": list(df.columns),
                "preview": df.head(20).to_csv(index=False),
                "stats": df.describe().to_json()
            }, ensure_ascii=False)
        return '{"error": "Unknown function"}'
 
    def analyze(self, file_path: str, question: str) -> dict:
        """
        ファイルを分析し、構造化された結果を返す
 
        Args:
            file_path: 分析対象のCSV/Excelファイルパス
            question: 分析に関する質問や指示
 
        Returns:
            dict: 構造化された分析結果
        """
        # Phase 1: Function Calling でデータを取得
        config = types.GenerateContentConfig(
            tools=[self.function_tool, self.code_exec_tool],
            system_instruction="""あなたはデータ分析の専門家です。
以下の手順で分析を行ってください:
1. まずload_csv_fileでデータを読み込む
2. Code Executionでデータを分析・可視化する
3. 発見したインサイトを詳細に説明する"""
        )
 
        prompt = f"""以下のファイルを分析してください。
ファイル: {file_path}
質問: {question}
 
手順:
1. load_csv_file関数でデータを読み込む
2. Pythonコードでデータを分析・可視化する
3. 結果をまとめる"""
 
        # マルチターンで Function Call を処理
        contents = [types.Content(
            role="user",
            parts=[types.Part(text=prompt)]
        )]
 
        graphs = []
        max_turns = 5
 
        for _ in range(max_turns):
            response = self.client.models.generate_content(
                model=self.model,
                contents=contents,
                config=config
            )
 
            # Function Call があれば実行して結果を返す
            fn_calls = [
                p for p in response.candidates[0].content.parts
                if p.function_call
            ]
 
            if fn_calls:
                contents.append(response.candidates[0].content)
                fn_responses = []
                for fc in fn_calls:
                    result = self._handle_function_call(fc.function_call)
                    fn_responses.append(
                        types.Part(function_response=types.FunctionResponse(
                            name=fc.function_call.name,
                            response=json.loads(result)
                        ))
                    )
                contents.append(types.Content(
                    role="user", parts=fn_responses
                ))
                continue
 
            # グラフ画像を収集
            for part in response.candidates[0].content.parts:
                if part.inline_data:
                    graphs.append(part.inline_data)
 
            break
 
        # Phase 2: Structured Output で結果を構造化
        analysis_text = response.text if response.text else ""
 
        structured_response = self.client.models.generate_content(
            model=self.model,
            contents=f"""以下の分析結果を構造化してください:
 
{analysis_text}""",
            config=types.GenerateContentConfig(
                response_mime_type="application/json",
                response_schema=types.Schema(
                    type="OBJECT",
                    properties={
                        "summary": types.Schema(type="STRING"),
                        "key_metrics": types.Schema(
                            type="ARRAY",
                            items=types.Schema(
                                type="OBJECT",
                                properties={
                                    "name": types.Schema(type="STRING"),
                                    "value": types.Schema(type="STRING"),
                                    "trend": types.Schema(type="STRING")
                                }
                            )
                        ),
                        "insights": types.Schema(
                            type="ARRAY",
                            items=types.Schema(
                                type="OBJECT",
                                properties={
                                    "title": types.Schema(type="STRING"),
                                    "description": types.Schema(type="STRING"),
                                    "action_item": types.Schema(type="STRING")
                                }
                            )
                        )
                    },
                    required=["summary", "key_metrics", "insights"]
                )
            )
        )
 
        result = json.loads(structured_response.text)
        result["graphs"] = [
            {
                "mime_type": g.mime_type,
                "data": base64.b64encode(g.data).decode()
            }
            for g in graphs
        ]
 
        return result
 
    def save_graphs(self, result: dict, output_dir: str = "./output"):
        """分析結果のグラフを画像ファイルとして保存"""
        Path(output_dir).mkdir(parents=True, exist_ok=True)
        for i, graph in enumerate(result.get("graphs", [])):
            ext = graph["mime_type"].split("/")[-1]
            path = Path(output_dir) / f"chart_{i}.{ext}"
            with open(path, "wb") as f:
                f.write(base64.b64decode(graph["data"]))
            print(f"💾 グラフ保存: {path}")

使用例:

agent = DataAnalysisAgent(model="gemini-3-flash")
 
result = agent.analyze(
    file_path="sales_2026_q1.csv",
    question="四半期の売上トレンドを分析し、成長率の高い商品カテゴリを特定してください"
)
 
print("📊 概要:", result["summary"])
for metric in result["key_metrics"]:
    print(f"  {metric['name']}: {metric['value']} ({metric['trend']})")
 
for insight in result["insights"]:
    print(f"\n💡 {insight['title']}")
    print(f"   {insight['description']}")
    print(f"   → アクション: {insight['action_item']}")
 
agent.save_graphs(result, output_dir="./reports/charts")

応用パターン

パターン 1: マルチファイル比較分析

複数ファイルを同時に読み込み、比較分析を行うパターンです。

# 複数期間のデータを比較分析
result = agent.analyze(
    file_path="sales_q1.csv",  # メインファイル
    question="""Q1とQ2の売上データを比較分析してください。
    追加データ: sales_q2.csv
    比較ポイント:
    - 売上成長率
    - 商品カテゴリ別シェアの変化
    - 季節性パターンの検出"""
)

パターン 2: 定期レポート自動生成

cron ジョブやスケジューラーと組み合わせて、定期的にレポートを自動生成します。

import schedule
import time
from datetime import datetime
 
def daily_report():
    agent = DataAnalysisAgent()
    today = datetime.now().strftime("%Y-%m-%d")
 
    result = agent.analyze(
        file_path=f"/data/daily/{today}.csv",
        question="本日の売上サマリーと前日比を分析してください"
    )
 
    # Markdownレポートを生成
    report = f"# 日次レポート ({today})\n\n"
    report += f"## 概要\n{result['summary']}\n\n"
    report += "## 主要指標\n"
    for m in result["key_metrics"]:
        report += f"- **{m['name']}**: {m['value']} ({m['trend']})\n"
 
    with open(f"reports/{today}.md", "w") as f:
        f.write(report)
 
schedule.every().day.at("09:00").do(daily_report)

パターン 3: 異常検知パイプライン

Code Execution 内で scikit-learn を使った異常検知を実行します。

result = agent.analyze(
    file_path="sensor_data.csv",
    question="""このセンサーデータから異常値を検出してください。
    手法:
    - Isolation Forest で外れ値を検出
    - 時系列の移動平均からの乖離を計算
    - 異常スコアが高い上位10件をリストアップ
    結果はグラフ（異常値を赤でハイライト）付きでお願いします。"""
)

トラブルシューティング

よくあるエラーと対処法

1. Code Execution のタイムアウト（30秒制限）

大規模データを処理する場合、30 秒の実行時間制限に引っかかることがあります。

# ❌ 大規模データをそのまま処理
"100万行のCSVをすべて分析してください"
 
# ✅ サンプリングまたは集約後に処理
"データをランダムに10,000行サンプリングし、統計的に分析してください"

2. Function Calling と Code Execution の競合

両方のツールを同時に指定すると、モデルがどちらを使うか迷う場合があります。

# System Instruction で明確に指示する
system_instruction = """
データの読み込みには必ずload_csv_file関数を使用してください。
データの分析・計算・グラフ生成にはCode Executionを使用してください。
この2つの役割を混同しないでください。
"""

3. Structured Output のスキーマエラー

複雑なスキーマを指定すると、モデルが正確に準拠できないことがあります。

# ❌ 深くネストしたスキーマ
# ✅ フラットな構造を優先し、必要に応じて段階的に構造化

コスト・パフォーマンス考慮事項

トークンコストの最適化

操作	入力トークン目安	出力トークン目安	Flash コスト概算
CSV読み込み（1,000行）	~5,000	~500	~$0.001
統計分析 + コード実行	~2,000	~3,000	~$0.003
グラフ生成（Matplotlib）	~1,000	~2,000	~$0.002
Structured Output	~3,000	~1,000	~$0.002
合計（1回の分析）	~11,000	~6,500	~$0.008

パフォーマンス最適化のヒント

プレビュー → 詳細分析の 2 段階アプローチ: 最初に少量のデータで概要を把握し、必要な部分だけ詳細分析する
Context Caching の活用: 同じデータセットに対して複数の質問を行う場合、Context Caching で入力トークンを再利用する
Flash vs Pro の使い分け: 日常の分析は Flash（低コスト・高速）、複雑な推論が必要な場合のみ Pro を使用

# Context Caching を使った効率的な複数質問
from google.genai import types
 
cache = client.caches.create(
    model="gemini-3-flash",
    contents=[types.Content(
        role="user",
        parts=[types.Part(text=f"分析対象データ:\n{large_csv_data}")]
    )],
    config=types.CreateCachedContentConfig(
        display_name="sales-data-q1",
        ttl="3600s"
    )
)
 
# キャッシュを使って複数回質問（入力トークン削減）
for question in ["売上トレンドは？", "異常値はある？", "予測モデルを構築して"]:
    response = client.models.generate_content(
        model="gemini-3-flash",
        contents=question,
        config=types.GenerateContentConfig(
            cached_content=cache.name,
            tools=[types.Tool(code_execution=types.ToolCodeExecution)]
        )
    )

まとめ

ここではGemini API の Code Execution、Function Calling、Structured Output を組み合わせて、自然言語の指示だけで CSV/Excel データを自動分析する AI エージェントを構築しました。

重要なポイント:

Code Execution はサンドボックス内で Pandas/Matplotlib を実行でき、データ処理・可視化に最適
Function Calling で外部データソースへのアクセスを抽象化し、エージェントの対応範囲を拡張
Structured Output で分析結果を型安全な JSON として取得し、後続処理に渡しやすくする
3 つの機能を組み合わせることで、単独では実現できない高度な自動分析パイプラインが構築可能

次のステップとして、Gemini API の Streaming 機能でリアルタイムに分析進捗を表示したり、Interactions API を使って長時間実行のリサーチタスクを統合したり、Function Calling の本番設計パターンを参考にエージェントの信頼性を向上させることをおすすめします。