Zenn
Google Cloud JapanPublicationへの投稿
💈

Geminiへの構造化データの入出力制御

Tanaike
2024/12/03に公開
AI
Python
LLM
Gemini
tech

概要

Gemini を用いた少サイズデータを用いたコンテンツ生成に焦点を当て、構造化データ（CSV、JSON など）を効果的に活用する方法をPythonを用いて解説します。Gemini が持つ大規模トークン処理能力のメリットを活かした、より効率的なコンテンツ生成手法を提案します。

序論

AI が急速に進化する時代において、大量のデータを分析・活用する能力は極めて重要です。RAG 環境はこうしたタスクに理想的ですが、より少量のデータでコンテンツ生成を行う必要があるシナリオもあります。Gemini は、大量のトークンを処理できるため、有望な解決策を提供します。プロンプトとアップロードされたファイルを組み合わせることで、限られたデータでも効果的に活用できます。ただし、CSV や JSON などの構造化データ形式を扱う際には、AI が情報を正確に解釈・理解できることを保証する必要があります。ここでは、サンプルとして Python を使用してこれを達成するための実用的なアプローチを検討します。AI が構造化データに基づいて効果的にコンテンツを理解・生成するためにどのようにトレーニングできるかを具体例を用いて示します。

このレポートはこちらを日本語として作成しました。

プロセス

プロセスは次の通りです。

  1. データ準備: プロンプトと必要なデータを含む CSV ファイルを用意します。
  2. スキーマ作成:
    A. CSV 直接使用: CSV スキーマを生成し、CSV ファイル内のデータの構造と型を定義します。
    B. CSV から JSON への変換: CSV データを JSON 形式で使用する場合、JSON スキーマを作成し、JSON データの構造と型を記述します。
  3. 出力スキーマ定義: 出力の構造と型を指定するための JSON スキーマを作成します。このスキーマは、プロンプト内、または response_schema パラメータを使用して、Gemini の生成プロセスをガイドすることができます。
  4. コンテンツ生成: Gemini を使って用意した入力データと定義したプロンプトを使用して、目的のコンテンツを生成します。生成されたコンテンツは、指定された出力スキーマに準拠します。
  5. 結果の返却: 生成されたコンテンツが最終的な出力として返されます。

使用方法

1. API キーの準備

https://ai.google.dev/gemini-api/docs/api-key にアクセスし、API キーを作成してください。その際、API コンソールでGenerative Language APIを有効にしてください。この API キーは、以下のスクリプトで使用されます。

この公式ドキュメントもご参照ください。Ref.

API キーを使わない場合は、これをスキップしてください。

2. サンプルデータ

ここで使用する CSV のサンプルデータは次の通りです。

ここでは、上記のサンプルデータを入力する CSV データとして使用します。画像は Google スプレッドシートを示していますが、実際のテストでは、このスプレッドシートから変換された CSV データを使用します。ここでは、CSV ファイルのファイル名はsample.csvとします。

サンプルデータは、e-Stat（日本の統計が閲覧できる政府統計ポータルサイト）の「A」、「B」、「C」列から取得したものです。これらの列は、それぞれ年、地域（日本のすべての都道府県）、および人口を表しています。画像は「北海道」のみを示していますが、実際のデータにはすべての都道府県が含まれています。データは2,303行3列からなり、以下で紹介するスクリプトで CSV 形式で使用されます。

このレポートでは、サンプルとして Gemini へ上記のデータを使って次のような要求を与えます。

  • 将来人口が最も増加する 3 つの地域を増加順に予測し、それらの地域名、増加の詳細な理由を出力する。
  • それぞれの地域の特徴を考慮して人口増加を維持するための対策を返す。
  • 人口増加を維持するための対策の有無に応じて、現在の人口と 50 年後の人口を予測する。

3. メインとなるスクリプト

これは Python のスクリプトです。

これは、次のセクションで使用するサンプルスクリプトを実行するためのメインクラスです。以下のスクリプトを含むGenerateContent.pyというファイルを作成してください。次のセクションで使用するサンプルスクリプトでは、このスクリプトをimport GenerateContentとして使用します。

import google.generativeai as genai
import io
import json
import requests
import time


class Main:

    def __init__(self):
        self.genai = None
        self.model = None
        self.api_key = None

    def run(self, object):
        self.api_key = object["api_key"]
        self._setInstances(object)

        print("Get file...")
        file = self._uploadFile(object["name"], object["data"])

        print("Generate content...")
        response = self.model.generate_content(
            [file, object["prompt"]], request_options={"timeout": 600}
        )

        data = None
        try:
            data = json.loads(response.text)
        except json.JSONDecodeError:
            data = response.text
        return data

    def _setInstances(self, object):
        genai.configure(api_key=self.api_key)
        generation_config = {"response_mime_type": "application/json"}
        if "response_schema" in object:
            generation_config["response_schema"] = object["response_schema"]
        self.genai = genai
        self.model = genai.GenerativeModel(
            model_name="gemini-1.5-flash-002", # or gemini-1.5-pro-002
            generation_config=generation_config,
        )

    def _uploadFile(self, name, text):
        file = None
        try:
            file = genai.get_file(f"files/{name}")
        except:
            requests.post(
                f"https://generativelanguage.googleapis.com/upload/v1beta/files?uploadType=multipart&key={self.api_key}",
                files={
                    "data": (
                        "metadata",
                        json.dumps(
                            {
                                "file": {
                                    "mimeType": "text/plain",
                                    "name": f"files/{name}",
                                }
                            }
                        ),
                        "application/json",
                    ),
                    "file": ("file", io.StringIO(text), "text/plain"),
                },
            )
            time.sleep(2)
            file = genai.get_file(f"files/{name}")
            print(f"File was uploaded.")
        while file.state.name == "PROCESSING":
            print(".", end="")
            time.sleep(10)
            file = genai.get_file(file.name)
        if file.state.name == "FAILED":
            raise ValueError(file.state.name)
        return file

サンプル1

このサンプルでは CSV データを直接使用します。Gemini に CSV データを理解させるために、CSV スキーマを使用しました。また、結果を JSON データとしてエクスポートするために、JSON スキーマを使用しました。これらは、次のスクリプトのcsvSchemaとjsonSchemaとして見ることができます。プロンプトはpromptとして見ることができます。

関数createDataは、生の CSV データを返します。

このスクリプトでは、出力用の JSON スキーマがプロンプトで使用されています。

import GenerateContent
import json


api_key = "###"  # Please set your API key.
filename = "sample.csv"  # Please set your CSV file with the path.


def createPrompt():
    csvSchema = {
        "description": 'Order of "fields" is the order of columns of CSV data.\n"name" is the column name.\n"type" is the type of value in the column.',
        "fields": [
            {"name": "Year", "type": "number"},
            {"name": "Region", "type": "string"},
            {"name": "Population", "type": "number"},
        ],
    }
    jsonSchema = {
        "description": "JSON schema for outputting the result.",
        "type": "array",
        "items": {
            "type": "object",
            "properties": {
                "region": {"description": "Region name.", "type": "string"},
                "reason": {
                    "description": "Reasons for the population increase.",
                    "type": "string",
                },
                "measures": {
                    "description": "Details of measures to stop the population increase.",
                    "type": "string",
                },
                "currentPopulation": {
                    "description": "Current population.",
                    "type": "number",
                },
                "futurePopulationWithMeasures": {
                    "description": "Future population after 50 years with measures to keep the population increasing.",
                    "type": "number",
                },
                "futurePopulationWithoutMeasures": {
                    "description": "Future population after 50 years without measures to keep the population increasing.",
                    "type": "number",
                },
            },
            "required": [
                "region",
                "reason",
                "measures",
                "currentPopulation",
                "futurePopulationWithMeasures",
                "futurePopulationWithoutMeasures",
            ],
        },
    }
    prompt = "\n".join(
        [
            "Run the following steps.",
            '1. Read the CSV data in the following text file. The CSV schema of this data is "CSVSchema".',
            f"<CSVSchema>{json.dumps(csvSchema)}</CSVSchema>",
            "2. Using the data collected and your knowledge, predict 3 regions that will have the largest increase in population in the future in the order of increase. Return the region name, detailed reasons for the increase, and measures to keep the population increasing by considering the features of the region. Also, return the current population and the population 50 years later predicted by you with and without measures to keep the population increasing.",
            '3. Return the result by following "JSONSchema".',
            f"<JSONSchema>{json.dumps(jsonSchema)}</JSONSchema>",
        ]
    )
    return prompt


def createData(filename):
    return open(filename, "r").read()


data = createData(filename)
prompt = createPrompt()
object = {
    "api_key": api_key,
    "name": "sample-name-1a",
    "data": data,
    "prompt": prompt,
}
res = GenerateContent.Main().run(object)
print(res)

この場合、出力値を設定するための JSON スキーマは、下記のようにresponse_schemaとして使用することも可能です。

import GenerateContent
import json


api_key = "###"  # Please set your API key.
filename = "sample.csv"  # Please set your CSV file with the path.


def createPrompt():
    csvSchema = {
        "description": 'Order of "fields" is the order of columns of CSV data.\n"name" is the column name.\n"type" is the type of value in the column.',
        "fields": [
            {"name": "Year", "type": "number"},
            {"name": "Region", "type": "string"},
            {"name": "Population", "type": "number"},
        ],
    }
    prompt = "\n".join(
        [
            "Run the following steps.",
            '1. Read the CSV data in the following text file. The CSV schema of this data is "CSVSchema".',
            f"<CSVSchema>{json.dumps(csvSchema)}</CSVSchema>",
            "2. Using the data collected and your knowledge, predict 3 regions that will have the largest increase in population in the future in the order of increase. Return the region name, detailed reasons for the increase, and measures to keep the population increasing by considering the features of the region. Also, return the current population and the population 50 years later predicted by you with and without measures to keep the population increasing.",
        ]
    )
    return prompt


def createData(filename):
    return open(filename, "r").read()


jsonSchema = {
    "description": "JSON schema for outputting the result.",
    "type": "array",
    "items": {
        "type": "object",
        "properties": {
            "region": {"description": "Region name.", "type": "string"},
            "reason": {
                "description": "Reasons for the population increase.",
                "type": "string",
            },
            "measures": {
                "description": "Details of measures to stop the population increase.",
                "type": "string",
            },
            "currentPopulation": {
                "description": "Current population.",
                "type": "number",
            },
            "futurePopulationWithMeasures": {
                "description": "Future population after 50 years with measures to keep the population increasing.",
                "type": "number",
            },
            "futurePopulationWithoutMeasures": {
                "description": "Future population after 50 years without measures to keep the population increasing.",
                "type": "number",
            },
        },
        "required": [
            "region",
            "reason",
            "measures",
            "currentPopulation",
            "futurePopulationWithMeasures",
            "futurePopulationWithoutMeasures",
        ],
    },
}
data = createData(filename)
prompt = createPrompt()
object = {
    "api_key": api_key,
    "name": "sample-name-1b",
    "data": data,
    "prompt": prompt,
    "response_schema": jsonSchema,
}
res = GenerateContent.Main().run(object)
print(res)

サンプル2

このサンプルでは、CSV データを JSON データに変換してから使用します。Gemini に JSON データを理解させるために、JSON スキーマを使用しました。また、結果を JSON データとしてエクスポートするために、JSON スキーマを使用しました。これらは、次のスクリプトのjsonSchema1とjsonSchema2として見ることができます。プロンプトはpromptとして見ることができます。

関数createDataは、CSV データから変換された JSON データを以下のように返します。

[
  {"region": "Hokkaido", "populations": [{"year": "1975", "population": "5338206"} ,,,]},
  {"region": "Aomori", "populations": [{"year": "1975", "population": "1468646"} ,,,]},
  {"region": "Iwate", "populations": [{"year": "1975", "population": "1385563"} ,,,]},
  ,
  ,
  ,
]

次のスクリプトは JSON スキーマをプロンプト内で使用しています。

import GenerateContent
import csv
import json


api_key = "###"  # Please set your API key.
filename = "sample.csv"  # Please set your CSV file with the path.


def createPrompt():
    jsonSchema1 = {
        "description": 'JSON schema of the inputted value. The filename is "blobName@sample.txt".',
        "type": "array",
        "items": {
            "type": "object",
            "properties": {
                "region": {"description": "Region name.", "type": "string"},
                "populations": {
                    "description": "Populations for each year.",
                    "type": "array",
                    "items": {
                        "type": "object",
                        "properties": {
                            "year": {"type": "number", "description": "Year."},
                            "population": {
                                "type": "number",
                                "description": "Population.",
                            },
                        },
                        "required": ["year", "populations"],
                    },
                },
            },
            "required": ["region", "populations"],
        },
    }
    jsonSchema2 = {
        "description": "JSON schema for outputting the result.",
        "type": "array",
        "items": {
            "type": "object",
            "properties": {
                "region": {"description": "Region name.", "type": "string"},
                "reason": {
                    "description": "Reasons for the population increase.",
                    "type": "string",
                },
                "measures": {
                    "description": "Details of measures to stop the population increase.",
                    "type": "string",
                },
                "currentPopulation": {
                    "description": "Current population.",
                    "type": "number",
                },
                "futurePopulationWithMeasures": {
                    "description": "Future population after 50 years with measures to keep the population increasing.",
                    "type": "number",
                },
                "futurePopulationWithoutMeasures": {
                    "description": "Future population after 50 years without measures to keep the population increasing.",
                    "type": "number",
                },
            },
            "required": [
                "region",
                "reason",
                "measures",
                "currentPopulation",
                "futurePopulationWithMeasures",
                "futurePopulationWithoutMeasures",
            ],
        },
    }
    prompt = "\n".join(
        [
            "Run the following steps.",
            '1. Read the JSON data in the following text file. The JSON schema of this data is "JSONSchema1".',
            f"<JSONSchema1>{json.dumps(jsonSchema1)}</JSONSchema1>",
            "2. Using the data collected and your knowledge, predict 3 regions that will have the largest increase in population in the future in the order of increase. Return the region name, detailed reasons for the increase, and measures to keep the population increasing by considering the features of the region. Also, return the current population and the population 50 years later predicted by you with and without measures to keep the population increasing.",
            '3. Return the result by following "JSONSchema2".',
            f"<JSONSchema2>{json.dumps(jsonSchema2)}</JSONSchema2>",
        ]
    )
    return prompt


def createData(filename):
    ar = list(csv.reader(open(filename, "r"), delimiter=","))[1:]
    obj = {}
    for r in ar:
        year, region, population = r
        v = {"year": year, "population": population}
        obj[region] = (obj[region] + [v]) if region in obj else [v]
    arr = [{"region": k, "populations": v} for (k, v) in obj.items()]
    return json.dumps(arr)


data = createData(filename)
prompt = createPrompt()
object = {
    "api_key": api_key,
    "name": "sample-name-2a",
    "data": data,
    "prompt": prompt,
}
res = GenerateContent.Main().run(object)
print(res)

この場合でも出力値を設定するための JSON スキーマは、下記のようにresponse_schemaとして使用することができます。

import GenerateContent
import csv
import json


api_key = "###"  # Please set your API key.
filename = "sample.csv"  # Please set your CSV file with the path.

def createPrompt():
    jsonSchema1 = {
        "description": 'JSON schema of the inputted value. The filename is "blobName@sample.txt".',
        "type": "array",
        "items": {
            "type": "object",
            "properties": {
                "region": {"description": "Region name.", "type": "string"},
                "populations": {
                    "description": "Populations for each year.",
                    "type": "array",
                    "items": {
                        "type": "object",
                        "properties": {
                            "year": {"type": "number", "description": "Year."},
                            "population": {
                                "type": "number",
                                "description": "Population.",
                            },
                        },
                        "required": ["year", "populations"],
                    },
                },
            },
            "required": ["region", "populations"],
        },
    }
    prompt = "\n".join(
        [
            "Run the following steps.",
            '1. Read the JSON data in the following text file. The JSON schema of this data is "JSONSchema1".',
            f"<JSONSchema1>{json.dumps(jsonSchema1)}</JSONSchema1>",
            "2. Using the data collected and your knowledge, predict 3 regions that will have the largest increase in population in the future in the order of increase. Return the region name, detailed reasons for the increase, and measures to keep the population increasing by considering the features of the region. Also, return the current population and the population 50 years later predicted by you with and without measures to keep the population increasing.",
        ]
    )
    return prompt

def createData(filename):
    ar = list(csv.reader(open(filename, "r"), delimiter=","))[1:]
    obj = {}
    for r in ar:
        year, region, population = r
        v = {"year": year, "population": population}
        obj[region] = (obj[region] + [v]) if region in obj else [v]
    arr = [{"region": k, "populations": v} for (k, v) in obj.items()]
    return json.dumps(arr)

jsonSchema2 = {
    "description": "JSON schema for outputting the result.",
    "type": "array",
    "items": {
        "type": "object",
        "properties": {
            "region": {"description": "Region name.", "type": "string"},
            "reason": {
                "description": "Reasons for the population increase.",
                "type": "string",
            },
            "measures": {
                "description": "Details of measures to stop the population increase.",
                "type": "string",
            },
            "currentPopulation": {
                "description": "Current population.",
                "type": "number",
            },
            "futurePopulationWithMeasures": {
                "description": "Future population after 50 years with measures to keep the population increasing.",
                "type": "number",
            },
            "futurePopulationWithoutMeasures": {
                "description": "Future population after 50 years without measures to keep the population increasing.",
                "type": "number",
            },
        },
        "required": [
            "region",
            "reason",
            "measures",
            "currentPopulation",
            "futurePopulationWithMeasures",
            "futurePopulationWithoutMeasures",
        ],
    },
}
data = createData(filename)
prompt = createPrompt()
object = {
    "api_key": api_key,
    "name": "sample-name-2b",
    "data": data,
    "prompt": prompt,
    "response_schema": jsonSchema2,
}
res = GenerateContent.Main().run(object)
print(res)

結果

得られた代表的な結果は次の通りです。

[
  {
    "region": "Tokyo",
    "reason": "Tokyo's robust economy, diverse job market, and well-established infrastructure continue to attract both domestic and international migrants.  Its status as a global hub for business and culture ensures ongoing population growth.",
    "measures": "Invest in affordable housing, improve public transportation, enhance green spaces and recreational facilities to improve quality of life, and continue promoting Tokyo as a global center for innovation and opportunity.",
    "currentPopulation": 14086000,
    "futurePopulationWithMeasures": 16000000,
    "futurePopulationWithoutMeasures": 15000000
  },
  {
    "region": "Osaka",
    "reason": "Osaka is a major economic center with a strong industrial base and a thriving service sector. Its vibrant culture and relatively lower cost of living compared to Tokyo attract individuals seeking opportunities.",
    "measures": "Focus on attracting skilled workers and entrepreneurs by offering tax incentives and streamlining business regulations. Improve affordable housing options and educational facilities. Promote Osaka's cultural attractions to attract tourists and residents.",
    "currentPopulation": 8763000,
    "futurePopulationWithMeasures": 10500000,
    "futurePopulationWithoutMeasures": 9500000
  },
  {
    "region": "Aichi",
    "reason": "Aichi Prefecture benefits from its position as a major manufacturing and automotive hub. This strong industrial base and associated employment opportunities fuel consistent population growth.",
    "measures": "Promote further diversification of the economy beyond automotive manufacturing to ensure long-term resilience. Invest in education and technology to attract highly skilled professionals. Develop sustainable infrastructure to enhance quality of life.",
    "currentPopulation": 7477000,
    "futurePopulationWithMeasures": 9000000,
    "futurePopulationWithoutMeasures": 8000000
  }
]

上記のサンプルスクリプトを実行した結果、以下のことがわかりました。

  • データ形式処理: Gemini は、対応するスキーマ（CSV スキーマと JSON スキーマ）を利用して、CSV と JSON の両方の構造化データ形式を正常に処理しました。
  • スキーマの有効性: 両方のスキーマアプローチは、Gemini が入力データを理解する上で有効であることが証明されました。
  • "reason" フィールドの可変性: "reason" フィールドの特定の値は、非ゼロの temperature を与えた場合、スクリプトの実行ごとにわずかに異なる可能性がありますが、他のフィールドの値は一定でした。
  • 地域予測の精度: 場合によっては、予測された地域が予想された結果と異なりました。ただし、この場合、主な焦点は、特定の地域予測にかかわらず、Gemini が入力データを正確に理解できたかどうかでした。

これらの結果は、Gemini が構造化データ形式を効果的に処理し、スキーマを利用して理解力を向上させることができることを示しています。また、入出力の制御ができるようになると、一度の API リクエストで複数の回答を得ることも可能になります。

応用

ここで紹介した方法を使用すると次のようなアプリケーションの作成が可能になります。詳細はそれぞれのレポートをご覧ください。

注記

  • ここでは、Python スクリプトが使用されていますが、ここで紹介したアプローチは Gemini API を利用するさまざまな言語に適用可能です。
  • データサイズは、Gemini API の最大トークン数によって決まりますのでご注意ください。
  • Google Cloud Champion Innovators Advent Calendar 2024 (3日目)

参考

Google Cloud Japan
Google Cloud JapanPublication

Google Cloud Japan のエンジニアを中心に情報発信をしている Publication です。Google Cloud サービスの技術情報を中心に公開しています。各記事の内容は個人の意見であり、企業を代表するものではございません。

Discussion