「JSONで答えて」だけでは足りない — AIでデータ前処理する仕組みからfunction callingを理解する

これまでデータを加工する仕組みをユーザーに提供するには、処理ごとに画面や操作項目を用意し、ユーザーにもその使い方を覚えてもらう必要がありました。

しかし、LLMが自然言語の指示をある程度正確に解釈できるようになったことで、ユーザーが処理名や操作手順を知らなくても、「取引先名を整えて」「日付の形式を統一して」と日本語で伝えるだけで、専門的な知識がそこまでなくても必要なデータ加工ができるようになってきました。

AIに日本語で指示して、CSVやExcelのデータを整形する仕組みを作るとします。ユーザーはコードを書く代わりに、次のように指示します。

売上データの取引先名から前後の空白と「株式会社」を取り除いてください。取引日をYYYY-MM-DD形式に統一し、顧客マスタと結合してcustomer_idを付けてください。

この指示を受けて、AIがあらかじめ用意されたデータ加工処理を選び、順番に実行する仕組みです。最初に思いつくのは、実行する関数や引数をJSONで返してもらう方法ではないでしょうか。

プロンプトだけで形式を指定すると、前置きやコードフェンスが付いたり、関数名や引数名が変わったりすることがあります。ただし、現在のLLM APIには JSON modeや Structured Outputs があるため、決まった形式のJSONを受け取りたいだけなら、JSONの形式を守らせるためにプロンプトを工夫したり、独自にパースしたりする必要はありません。

今回必要なのは、JSONを1つ受け取って終わる仕組みではありません。

CSVの内容を確認する
    ↓
文字列を整形する
    ↓
日付形式を統一する
    ↓
マスタと結合する

AIが関数を選び、アプリケーションが実行し、その結果を見て次の関数を選ぶ。このやり取りを実現するのがfunction callingです。

作りたいもの

あるクライアントでは、基幹システムから毎月出力される売上CSVを、分析やシステム連携に使える形へ整える必要がありました。

出力されるデータには、次のような問題があります。

• 取引先名の前後に空白が入っている
• 「株式会社サンプル」と「サンプル株式会社」が混在している
• 日付が2026/7/1や2026年07月01日など複数の形式になっている
• 売上データに社内の顧客IDが含まれていない
• 顧客マスタとの結合を毎月手作業で行っている

そこで、データサイエンティストやデータエンジニアなどが次の4つの処理をPythonで用意したとします。

1. CSVの列名やデータ例を確認する
2. 文字列を整形する
3. 日付形式を統一する
4. マスタデータと結合する

ユーザーは、関数名や処理手順を知る必要はありません。

sales.csvの取引先名から前後の空白と「株式会社」を除去してください。
取引日をYYYY-MM-DD形式に統一し、customer_master.csvと結合して、
customer_idを追加したprocessed_sales.csvを出力してください。

この指示をAIが解釈し、4つの関数を適切な順番で呼び出すことを目指します。

まずは普通のPython関数を作る

function callingを使う前に、実際のデータ処理を行う関数を作ります。

from typing import Literal

import pandas as pd


# LLMから指定できる文字列操作をあらかじめ限定する
TextOperation = Literal[
    "trim",
    "remove_corporate_designator",
    "normalize_width",
]


def inspect_csv(path: str) -> dict:
    """CSVの列名、型、先頭データを返す"""

    # CSVを読み込む
    df = pd.read_csv(path)

    # LLMが次の処理を判断できるように、
    # 列名・データ型・サンプル・行数を返す
    return {
        "columns": df.columns.tolist(),
        "dtypes": {
            column: str(dtype)
            for column, dtype in df.dtypes.items()
        },
        "sample": df.head(3).to_dict(orient="records"),
        "row_count": len(df),
    }


def normalize_text(
    input_path: str,
    output_path: str,
    column: str,
    operations: list[TextOperation],
) -> dict:
    """指定した列の文字列を整形する"""

    # 入力CSVを読み込む
    df = pd.read_csv(input_path)

    # 指定された列が存在するか確認する
    if column not in df.columns:
        raise ValueError(f"列が見つかりません: {column}")

    values = df[column].astype("string")

    # 指定された文字列操作を順番に適用する
    for operation in operations:
        if operation == "trim":
            # 前後の空白を削除する
            values = values.str.strip()

        elif operation == "remove_corporate_designator":
            # 「株式会社」を削除する
            values = (
                values
                .str.replace("株式会社", "", regex=False)
                .str.strip()
            )

        elif operation == "normalize_width":
            # 全角・半角などの表記をUnicode正規化で揃える
            values = values.str.normalize("NFKC")

        else:
            raise ValueError(f"未対応の操作です: {operation}")

    # 整形結果を元のDataFrameへ戻す
    df[column] = values

    # 処理後のCSVを保存する
    df.to_csv(output_path, index=False)

    return {
        "output_path": output_path,
        "row_count": len(df),
    }


def normalize_date(
    input_path: str,
    output_path: str,
    column: str,
    output_format: str,
) -> dict:
    """指定した列の日付形式を統一する"""

    # 入力CSVを読み込む
    df = pd.read_csv(input_path)

    # 指定された列が存在するか確認する
    if column not in df.columns:
        raise ValueError(f"列が見つかりません: {column}")

    # 「2026年7月1日」のような値を
    # pandasが解釈しやすい形式へ変換する
    values = (
        df[column]
        .astype("string")
        .str.replace("年", "-", regex=False)
        .str.replace("月", "-", regex=False)
        .str.replace("日", "", regex=False)
    )

    # 複数の日付形式を日付型として解析する
    parsed = pd.to_datetime(
        values,
        format="mixed",
        errors="coerce",
    )

    # 日付として解釈できなかった値を確認する
    invalid = df.loc[
        df[column].notna() & parsed.isna(),
        column,
    ]

    if not invalid.empty:
        raise ValueError(
            f"日付として解釈できない値があります: "
            f"{invalid.head(3).tolist()}"
        )

    # 指定された形式へ統一する
    df[column] = parsed.dt.strftime(output_format)

    # 処理後のCSVを保存する
    df.to_csv(output_path, index=False)

    return {
        "output_path": output_path,
        "row_count": len(df),
    }


def join_master(
    input_path: str,
    master_path: str,
    output_path: str,
    left_on: str,
    right_on: str,
    master_columns: list[str],
) -> dict:
    """入力データとマスタを結合する"""

    # 入力データとマスタデータを読み込む
    df = pd.read_csv(input_path)
    master = pd.read_csv(master_path)

    # 入力データ側に結合キーが存在するか確認する
    if left_on not in df.columns:
        raise ValueError(
            f"入力データに結合列がありません: {left_on}"
        )

    # マスタ側で必要になる列を整理する
    required_columns = [right_on, *master_columns]

    # 必要な列がマスタに存在するか確認する
    missing_columns = [
        column
        for column in required_columns
        if column not in master.columns
    ]

    if missing_columns:
        raise ValueError(
            f"マスタに必要な列がありません: {missing_columns}"
        )

    # 入力データを基準にマスタをLEFT JOINする
    # many_to_oneでマスタ側のキー重複も検出する
    result = df.merge(
        master[required_columns],
        how="left",
        left_on=left_on,
        right_on=right_on,
        validate="many_to_one",
        indicator=True,
    )

    # マスタと一致しなかった行数を数える
    unmatched_count = int(
        (result["_merge"] == "left_only").sum()
    )

    # 結合確認用の一時列を削除する
    result = result.drop(columns=["_merge"])

    # 結合結果をCSVとして保存する
    result.to_csv(output_path, index=False)

    return {
        "output_path": output_path,
        "row_count": len(result),
        "unmatched_count": unmatched_count,
    }

ここまでは、function callingとは関係のない、表データを操作するためのPandasを使用した普通のPythonの処理です。データ処理自体はデータサイエンティストやエンジニアが実装します。LLMに任せるのは、この中から何を選び、どの引数で呼ぶかという部分です。

LLMに直接Pythonコードを書かせればよいのではないか

現在のAIであれば、先ほどの日本語からpandasのコードを直接生成することもできます。一度だけ行う操作や、エンジニアが内容を確認してから実行する用途であれば、それでも十分かと思います。

しかし、ユーザーが定期的に使う業務のための処理として考えると、生成されたコードをそのまま実行すると問題が発生しかねません。

• 毎回異なるコードが生成される可能性がある
• 想定していないファイルや列を変更する可能性がある
• 実行前にコードの安全性を確認する必要がある
• テスト済みの処理だけを使わせることが難しい
• 同じ指示でも同じ結果になるとは限らない

Function callingでは、LLMに自由なコードを書かせません。

ユーザーの日本語による指示
        ↓
LLMが指示を解釈する
        ↓
用意された前処理関数を選ぶ
        ↓
アプリケーションが検証して実行する

自然言語の柔軟さを利用しながら、実行する処理は特定の自分達が用意したコードで管理できます。

function callingとは

function callingは、利用可能な関数の名前、説明、引数の構造をLLMに渡し、ユーザーの指示に応じて「どの関数を、どの引数で呼ぶか」をLLMに選ばせる仕組みです。

基本的な流れは次のとおりです。

1. アプリケーションが関数の一覧をLLMへ渡す
2. LLMが関数名と引数を返す
3. アプリケーションが引数を検証し、関数を実行する
4. 実行結果をLLMへ返す
5. LLMが次の関数を選ぶか、処理の完了を伝える

重要なのは、LLM自身がPython関数を直接実行するわけではないことです。LLMが返すのは「この関数を、この引数で呼びたい」という、あくまで要求であり、実行するのはアプリケーション側です。

OpenAIではfunction callingまたはtool calling、Anthropicではtool useと呼ばれています。API上の形式は異なりますが、モデルがツールを選び、アプリケーションが実行結果を返す流れは共通しているのです。

関数の引数をJSON Schemaで定義する

function callingでは、関数名や説明に加えて、引数の構造をJSON Schemaで定義します。

この考え方はOpenAIとAnthropicで共通していますが、OpenAIはparameters、Anthropicはinput_schemaを使うなど、API上の形式は異なります。以下ではOpenAI Responses APIを使います。

まず、Pydanticで関数の引数を定義します。

from pydantic import BaseModel, ConfigDict, Field


class StrictArgs(BaseModel):
    model_config = ConfigDict(
        extra="forbid",
        strict=True,
    )


class InspectCSVArgs(StrictArgs):
    path: str = Field(
        description="確認するCSVファイルのパス"
    )


class NormalizeTextArgs(StrictArgs):
    input_path: str
    output_path: str
    column: str
    operations: list[TextOperation]


class NormalizeDateArgs(StrictArgs):
    input_path: str
    output_path: str
    column: str
    output_format: Literal["%Y-%m-%d"]


class JoinMasterArgs(StrictArgs):
    input_path: str
    master_path: str
    output_path: str
    left_on: str = Field(
        description="入力データ側の結合列"
    )
    right_on: str = Field(
        description="マスタ側の結合列"
    )
    master_columns: list[str] = Field(
        description="マスタから追加する列"
    )

extra="forbid"は定義していない引数を拒否し、strict=Trueは暗黙の型変換を抑えます。

関数名、説明、引数モデル、実際のPython関数をまとめて登録します。

TOOL_SPECS = {
    "inspect_csv": {
        "description": (
            "CSVの列名、型、行数、先頭データを確認する。"
            "列名が不明な場合は加工前に使う"
        ),
        "args_model": InspectCSVArgs,
        "function": inspect_csv,
    },
    "normalize_text": {
        "description": (
            "指定した文字列列に対して、"
            "空白除去、法人格除去、文字幅の統一を行う"
        ),
        "args_model": NormalizeTextArgs,
        "function": normalize_text,
    },
    "normalize_date": {
        "description": (
            "指定した日付列をYYYY-MM-DD形式に統一する"
        ),
        "args_model": NormalizeDateArgs,
        "function": normalize_date,
    },
    "join_master": {
        "description": (
            "入力データとマスタを結合し、"
            "マスタから必要な列を追加する"
        ),
        "args_model": JoinMasterArgs,
        "function": join_master,
    },
}


tools = [
    {
        "type": "function",
        "name": name,
        "description": spec["description"],
        "parameters": spec["args_model"].model_json_schema(),
        "strict": True,
    }
    for name, spec in TOOL_SPECS.items()
]

OpenAIのStrict modeでは、各オブジェクトにadditionalProperties: falseを指定し、すべてのプロパティをrequiredに含める必要があります。strict: trueを指定すると、関数の引数がスキーマに従うよう制約されます。

ただし、保証されるのは引数の構造だけです。

• ファイルが存在するか
• 列名が正しいか
• マスタの結合キーが一意か
• 出力先を上書きしてよいか
• ユーザーがそのデータを操作できるか

このような内容の妥当性は、コード側で検証する必要があります。

関数を検証して実行する

LLMが返した関数名は、信頼できない外部入力として扱います。

eval()や無制限なgetattr()は使わず、登録済みの関数だけを実行します。

from pydantic import ValidationError


def execute_tool(
    name: str,
    arguments_json: str,
) -> dict:
    spec = TOOL_SPECS.get(name)

    if spec is None:
        return {
            "ok": False,
            "error": f"unknown function: {name}",
        }

    try:
        args = spec["args_model"].model_validate_json(
            arguments_json
        )

        result = spec["function"](
            **args.model_dump()
        )

        return {
            "ok": True,
            "result": result,
        }

    except ValidationError as error:
        return {
            "ok": False,
            "error": f"invalid arguments: {error}",
        }

    except Exception as error:
        return {
            "ok": False,
            "error": str(error),
        }

列が存在しない、日付を解釈できない、マスタのキーが重複しているといったエラーも、実行結果としてLLMへ返されます。LLMはエラーを見て別の引数を選べますが、必ず修正できるわけではありません。そのため、実行回数には上限を設けたほうがいいでしょう。

呼び出しループを作る

以下がOpenAI Responses APIを使った最小構成です。

import json
import os

from openai import OpenAI


client = OpenAI()
MAX_STEPS = 12


def run(user_instruction: str) -> str:
    input_items = [
        {
            "role": "user",
            "content": user_instruction,
        }
    ]

    for _ in range(MAX_STEPS):
        response = client.responses.create(
            model=os.environ["OPENAI_MODEL"],
            instructions=(
                "あなたはCSVの前処理を支援します。"
                "列名やデータが不明な場合は、"
                "加工前にinspect_csvを使ってください。"
                "各処理の出力ファイルを、"
                "次の処理の入力として使用してください。"
            ),
            input=input_items,
            tools=tools,
            parallel_tool_calls=False,
        )

        input_items += response.output

        calls = [
            item
            for item in response.output
            if item.type == "function_call"
        ]

        if not calls:
            return response.output_text

        for call in calls:
            result = execute_tool(
                name=call.name,
                arguments_json=call.arguments,
            )

            input_items.append(
                {
                    "type": "function_call_output",
                    "call_id": call.call_id,
                    "output": json.dumps(
                        result,
                        ensure_ascii=False,
                    ),
                }
            )

    raise RuntimeError(
        "function callingの実行回数が上限を超えました"
    )

Responses APIでは、返されたfunction_callを実行し、同じcall_idを持つfunction_call_outputとして結果を返します。parallel_tool_calls=Falseを指定すると、1回の応答に含まれるツール呼び出しを0件または1件に制限できます。

今回の指示であれば、次のような順序になります。

1. inspect_csv("sales.csv")
2. inspect_csv("customer_master.csv")
3. normalize_text(...)
4. normalize_date(...)
5. join_master(...)

最初の時点でLLMはCSVの中身を知りません。

inspect_csv()の結果を返すことで、実際の列名やデータ例を踏まえて次の処理を選べるようになります。この繰り返しがfunction callingの基本形です。

メリットとデメリット

また、関数が増えるほど、似た関数同士の選択が難しくなります。

例えば、大文字化、小文字化、空白除去を別々の関数にするのではなく、次のようにまとめられます。

OpenAIの公式ドキュメントでは、1ターンの開始時に利用可能にする関数を20件未満にすることが、あくまで柔らかい目安として示されています。関数定義自体も入力トークンとして消費されます。

一方ですべてを巨大な1関数にまとめると、引数の組み合わせが複雑になります。LLMが関数の役割を区別しやすい単位に分ける必要があります。

Structured Outputsとの違い

Structured Outputsとfunction callingは、どちらもJSON Schemaを利用できますが、用途が異なります。

OpenAIの公式ドキュメントでも、モデルの回答形式を構造化する場合はStructured Outputs、モデルとアプリケーションの機能を接続する場合はfunction callingと整理されています。

また、毎月必ず同じ列を同じ順番で処理するなら、function callingは必要ありません。

normalize_text(...)
normalize_date(...)
join_master(...)

と通常のPythonコードで固定した方が、速く、安く、安定します。

function callingが向いているのは、ユーザーによって対象ファイル、列、処理内容、処理順などが変わる場合です。固定できる部分までLLMに判断させる必要はありません。

まとめ

Function callingの本質は、LLMに処理を実行させることではありません。ユーザーの自然言語による曖昧な指示を解釈する部分をLLMに任せ、実際の処理の実行と検証は自分のコードに任せる仕組みです。

日本語によるデータ前処理の指示
        ↓
LLMが関数と引数を選ぶ
        ↓
アプリケーションが検証して実行する
        ↓
結果をLLMへ返す
        ↓
次の処理を選ぶか、完了を伝える

AIでデータ前処理を行うからといって、毎回LLMにpandasのコードを生成させる必要はありません。LLMには「何をするか」の解釈を任せる。
実際に「何をしてよいか」を決め、再現可能な形で実行するのは自分のコードです。

この役割分担が、function callingを使う最大の理由です。