GoogleColabでFunctionCallingをGeminiで実装する

はじめに

本記事では前回の記事で実装したRAGに加えて生成AI周りの技術であるFunctionCallingがGeminiAPIより提供されていたのでそちらを実装していきたいと思います。

前回の記事では大分県の食べ物について詳しく回答するモデルを作成しましたが本記事では大分県内の温泉の情報を追加して温泉とグルメどちらかもしくは両方を聞かれた際にモデルがプロンプトからどの情報元を参照するべきかを判断して適切な回答を生成することを目指します。

結論から申し上げますと温泉とグルメの情報をモデルがうまく切り分けられなかったので記事の前半は日本の各都道府県の天気のリアルタイム情報を取得して生成AIが処理を行うソースコードを記載します。

目次

• 参考にしたサイト

• リアルタイムお天気モデルの実装

• 大分県のグルメ・温泉モデルの実装

参考にしたサイト

リアルタイムお天気モデルの実装

まずは、プロンプトに都道府県と天気に関わる質問をするとその都道府県のリアルタイムの天気情報を取得したうえで質問に答えるように実装していきます。

なんと参考にしたサイトでほとんど完成してしまっているので必要な作業はリアルタイムの天気の情報のみです！

下準備

まずはマウントと必要なライブラリのマウントです。

from google.colab import drive
drive.mount('/content/drive')

!pip install --upgrade google-generativeai

import google.generativeai as genai
from google.colab import userdata
import time
import json
import re
import requests

APIkeyの準備

次に必要なAPIキーをインポートします。
本記事で利用するAPIキーはGoogleAPIキーとOpenWeatherMapのAPIキーです

持っていない方は以下よりAPIキーを取得してください！

• Google studio
• OpenWeatherMap

インポート後以下のコマンドを実行してください。

# APIキーを設定
genai.configure(api_key=userdata.get('GOOGLE_API_KEY')) # GoogleAPIキー
api_key = userdata.get('WEATHER_API_KEY')  # OpenWeatherMapのAPIキーを設定

geminiから都道府県の県庁所在地の緯度と経度を取得する

今回利用するOpenWeatherMapのAPIは緯度と経度からその土地の天気を返却するのでgeminiを用いて都道府県から緯度と経度を出力するようにお願いします。

def get_prefecture_data(prefecture_name):
    content = f'''
        <rule>
        あなたは
        {prefecture_name}
        の緯度と経度をJSON形式で返却するAIです。
        <example>には東京都と入力された場合に出力すべき文字を示している
        <example>に示した形式に従い、余計な文言を付け足すな
        </rule>
        <example>
        {{"東京都": {{"lat": 35.6895, "lon": 139.6917}}}}
        </example>
        '''

    model = genai.GenerativeModel("gemini-1.5-flash")
    response = model.generate_content(content)
    # 無料版のリクエスト制限回避
    time.sleep(4)
    response = response.text.replace("\n", "").rstrip()
    # 正規表現で {} に囲まれた部分を抽出
    match = re.search(r'\{.*\}', response)
    if match:
        json_string = match.group(0)  # 抽出されたJSON文字列

        # JSON文字列を辞書に変換
        try:
            response = json.loads(json_string)
        except json.JSONDecodeError as e:
            print(f"JSONデコードエラー: {e}")
    else:
        print("JSON形式のデータが見つかりません。")
    print("reselt:", response)
    return response

OpenWeatherMapよりその土地の天気情報を取得

先ほど定義した緯度経度情報を取得する関数を用いてその土地の天気情報を取得します。

import requests
def get_weather(prefecture_name:str):
  # 緯度経度情報の取得
    PREFECTURE_DATA = get_prefecture_data(prefecture_name)
    if prefecture_name not in PREFECTURE_DATA:
        return f"{prefecture_name} の天候データは取得できません。"

    lat = PREFECTURE_DATA[prefecture_name]["lat"]
    lon = PREFECTURE_DATA[prefecture_name]["lon"]
    url = f"http://api.openweathermap.org/data/2.5/weather?lat={lat}&lon={lon}&appid={api_key}&lang=ja&units=metric"

    try:
        response = requests.get(url)
        response.raise_for_status()  # ステータスコードがエラーの場合例外を発生
        data = response.json()

        # 天候情報の取得
        weather = data["weather"][0]["description"]
        temp = data["main"]["temp"]
        return f"{prefecture_name}の天候は「{weather}」、気温は{temp}°Cです。"

    except requests.exceptions.RequestException as e:
        return f"天候情報の取得に失敗しました: {e}"

試しに生成AIから緯度経度情報を取得後、APIを経由して天気情報を入手

ここまで定義した関数を用いてある都道府県の天気情報を取得できるか試します。

if __name__ == "__main__":
  prefecture_name = "福岡県"  # 都道府県名を指定
  weather_info = get_weather(prefecture_name)
  print(weather_info)

• 出力

reselt: {'福岡県': {'lat': 33.59, 'lon': 130.4}}
福岡県の天候は「曇りがち」、気温は9.94°Cです。

正しそうです。

ここからがFunctionCalling

さて、ここからが本題のFunctionCallingの実装に入ります。

まずはライブラリのインストールです

!pip install -U -q google-generativeai
import google.ai.generativelanguage as glm

呼び出される関数を定義する

ここでは、プロンプトをモデルが受けて参照するべき関数があるか参照するための変数を初期化します。

# 関数を辞書で管理
functions = {
    "get_weather": get_weather,
}

モデルの準備

ここでは改めてFunctionCalling用のモデルを初期化します。

model = genai.GenerativeModel(
    model_name="gemini-1.5-flash-latest",
    tools=functions.values(),
)

謎の呪文を打ち込む

勉強不足により、正直なところ何をしているのかよくわかっていません。
有識者の方、コメントで教えていただけますと幸いです！

ここについては学習し次第更新しようと思います。

# 関数呼び出し
def call_function(function_call, functions):
    function_name = function_call.name
    function_args = function_call.args
    return functions[function_name](**function_args)

# 関数レスポンス群の生成
function_responses = []
for part in response.parts:
    if part.function_call:
        # 関数呼び出しの実行
        result = call_function(part.function_call, functions)

        function_response = glm.Part(function_response=glm.FunctionResponse(
            name=part.function_call.name,
            response={"result": result}
        ))
        function_responses.append(function_response)
print(function_responses)

# 会話履歴の作成
messages = [
    {'role':'user',
     'parts': [prompt]},
    {'role':'model',
     'parts': response.candidates[0].content.parts},
    {'role':'user',
     'parts': function_responses}
]

# 質問応答
response = model.generate_content(messages)
print(response.text)

• 出力

東京都の天候は薄い雲で、気温は12.33°Cです。大阪府の天候は曇りがちで、気温は11.89°Cです。福岡県の天候は曇りがちで、気温は10.99°Cです。

回答は確かにリアルタイムの情報が反映されたものになりましたが、深い思考はできないようです。
実装面といい回答面といいい何ともしまりの悪い結果となりましたがひとまずFunctionCallingの完成です！

大分県のグルメ・温泉モデルの実装

さて、やっと本記事の目的である大分県モデルについてです！
前のセクションとやることは参照する情報源が変わらないのでこのセクションでは割愛させていただいてソースコードをどんどん載せていきます！

情報はRAGを用いて実装しており、前回の記事で紹介しておりますので気になる方はご参照してください！

ライブラリの準備

%pip install --upgrade google-generativeai
%pip install -U -q google-generativeai
%pip install transformers accelerate sentencepiece
%pip install bitsandbytes
%pip install unstructured
%pip install selenium
%pip install sentence-transformers
%pip install chromadb
%pip install langchain
%pip install langchain-community
%pip install langchain-google-genai
%pip install sentence-transformers
%pip install langchain langchain_community
%pip install docling
%pip install faiss-cpu
%pip install unstructured
%pip install langchain-google-genai
%pip install langchain-experimental

import pathlib
import time
import json
import sys
import os
import langchain
import transformers
import torch
import bitsandbytes
import google.generativeai as genai
from google.colab import userdata
import langchain.llms
import langchain.prompts
import langchain.text_splitter
import langchain.embeddings
import langchain.vectorstores
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_experimental.text_splitter import SemanticChunker
import langchain_community.document_loaders
from langchain_community.vectorstores.faiss import FAISS
from langchain_core.prompts import PromptTemplate
from langchain_community.embeddings.huggingface import HuggingFaceEmbeddings
from langchain_community.document_loaders import DirectoryLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_google_genai import ChatGoogleGenerativeAI
from langchain.chains import RetrievalQA
from docling.document_converter import DocumentConverter

温泉とグルメのRAGを実装

# プロンプトから温泉に関連するデータに関するデータのみ抽出
def get_spa_data(prompt):
    content = f'''
        <rule>
        あなたは文章から温泉についての特徴を単語で答えるするAIです。
        <example>にはhumanにより入力された場合に出力すべき文字をassistantにより示している。
        <example>に示した形式に従い、余計な文言を付け足すな。
        </rule>
        <item>
        {prompt}
        </item>
        <example>
        human: 私は今、温泉を探しています。その温泉の特徴は露天風呂付、家族風呂付です。
        assistant: [露天風呂・家族風呂]
        </example>
        '''
    model = genai.GenerativeModel("gemini-1.5-flash")
    response = model.generate_content(content)
    # 無料版のリクエスト制限回避
    time.sleep(4)
    response = response.text.replace("\n", "").rstrip()

    print("reselt:", response)
    return response

# 温泉データ抽出
def get_spa_info_1(content:str):
    content = get_gourmets_data(content)
    spa_urls = [
        'https://www.visit-oita.jp/spa/',
    ]

    # ウェブページの内容を読込む
    loader = langchain_community.document_loaders.UnstructuredURLLoader(
        urls=spa_urls
    )
    spa_docs = loader.load()

    # 読込した内容を分割する（再帰的に分割）
    text_splitter = langchain.text_splitter.RecursiveCharacterTextSplitter(
        chunk_size=100,
        chunk_overlap=10,
    )
    # ベクトル化する準備
    spa_docs = text_splitter.split_documents(spa_docs)

    embedding = langchain.embeddings.HuggingFaceEmbeddings(
        model_name="intfloat/multilingual-e5-base"
    )
    # 読込した内容を保存
    spa_vectorstore = langchain.vectorstores.Chroma.from_documents(
        documents=spa_docs,
        embedding=embedding
    )

    query = content

    spa_docs = spa_vectorstore.similarity_search(query=query, k=10)

    for index, doc in enumerate(spa_docs):
        print("%d:" % (index + 1))
        print(doc.page_content)
    return spa_vectorstore

# RAGを用いた食べ物の検索
def get_spa(query:str):
    # プロンプトを準備
    template = """
    <bos><start_of_turn>system
    次の文脈を使用して、最後の質問に答えてください。
    {context}
    <end_of_turn><start_of_turn>user
    {query}
    <end_of_turn><start_of_turn>model
    """
    prompt = langchain.prompts.PromptTemplate.from_template(template)

    # チェーンを準備
    chain = (
        prompt
        | llm
    )
    # 検索する
    gourmets_vectorstore_1 = get_spa_info_1(query)
    docs = gourmets_vectorstore_1.similarity_search(query=query, k=10)

    content = "\n".join([f"Content:\n{doc.page_content}" for doc in docs])
    response = chain.invoke({
        'query': query,
        'context': content,
    })

    time.sleep(4)
    print("------------------------------------------------------------------")
    # print(response.content)

    return response.content

def get_gourmets_data(prompt):
    content = f'''
        <rule>
        あなたは文章から食べ物についての特徴を単語で答えるするAIです。
        <example>にはhumanにより入力された場合に出力すべき文字をassistantにより示している。
        <example>に示した形式に従い、余計な文言を付け足すな。
        </rule>
        <item>
        {prompt}
        </item>
        <example>
        human: あなたは今、大分県の魚を使った食べ物を探しています。3つおいしい新鮮な食べ物を紹介してください
        assistant: [魚・新鮮]
        </example>
        '''
    model = genai.GenerativeModel("gemini-1.5-flash")
    response = model.generate_content(content)
    # 無料版のリクエスト制限回避
    time.sleep(4)
    response = response.text.replace("\n", "").rstrip()

    print("reselt:", response)
    return response

# グルメデータ抽出
def get_gourmets_info(content:str):
    #プロンプトから食べ物に関する情報のみを取得する
    content = get_gourmets_data(content)


    gourmets_urls = [
        'https://www.visit-oita.jp/gourmets/#genre01',
    ]

    # ウェブページの内容を読込む
    loader = langchain_community.document_loaders.UnstructuredURLLoader(
        urls=gourmets_urls
    )
    gourmets_docs = loader.load()
    print("あ＝", gourmets_docs)

    # 読込した内容を分割する（再帰的に分割）
    text_splitter = langchain.text_splitter.RecursiveCharacterTextSplitter(
        chunk_size=50,
        chunk_overlap=10,
    )
    # ベクトル化する準備
    gourmets_docs = text_splitter.split_documents(gourmets_docs)

    embedding = langchain.embeddings.HuggingFaceEmbeddings(
        model_name="intfloat/multilingual-e5-base"
    )
    # 読込した内容を保存
    gourmets_vectorstore_1 = langchain.vectorstores.Chroma.from_documents(
        documents=gourmets_docs,
        embedding=embedding
    )

    query = content

    gourmets_docs = gourmets_vectorstore_1.similarity_search(query=query, k=10)

    for index, doc in enumerate(gourmets_docs):
        print("%d:" % (index + 1))
        print(doc.page_content)
    return gourmets_vectorstore_1

# RAGを用いた食べ物の検索
def get_food(query:str):
    # プロンプトを準備
    template = """
    <bos><start_of_turn>system
    次の文脈を使用して、最後の質問に答えてください。
    {context}
    <end_of_turn><start_of_turn>user
    {query}
    <end_of_turn><start_of_turn>model
    """
    prompt = langchain.prompts.PromptTemplate.from_template(template)

    # チェーンを準備
    chain = (
        prompt
        | llm
    )
    # 検索する
    gourmets_vectorstore_1 = get_gourmets_info(query)
    docs = gourmets_vectorstore_1.similarity_search(query=query, k=10)

    content = "\n".join([f"Content:\n{doc.page_content}" for doc in docs])
    response = chain.invoke({
        'query': query,
        'context': content,
    })

    time.sleep(4)
    print("------------------------------------------------------------------")
    # print(response.content)

    return response.content

質問をする

さて、早くも準備完了したので実行してみます。

if __name__ == "__main__":
  query = "あなたは今、大分市の魚を使った食べ物を探しています。おいしい新鮮な食べ物を3つ紹介してください。"
  question_info = get_food(query)
  print(question_info)

• 出力

大分市の魚を使った料理の情報がテキストに不足しているため、具体的な料理名を3つ挙げることはできません。テキストからは、大分県で生産されたブリ（かぼすを餌に育てたもの）が紹介されていますが、それが大分市で食べられるかどうかは不明です。

そのため、大分市の魚を使ったおいしい新鮮な食べ物を3つ紹介する代わりに、テキストの情報と一般的な知識を組み合わせた推測に基づいた提案をさせていただきます。  大分市で新鮮な魚料理を楽しむには、以下の3つのアプローチが考えられます。


1. **かぼすブリを使った料理:** テキストで紹介されているかぼすを餌に育てたブリは、大分県産で抗酸化作用により鮮度が保たれているとあります。大分市内の魚料理店や、スーパーマーケットなどでこのブリを使った刺身、照り焼き、しゃぶしゃぶなどを探してみるのが良いでしょう。  新鮮なブリの旨味とかぼすの爽やかな酸味の組み合わせは絶品です。


2. **大分市近海で獲れた旬の魚料理:** 大分市は海に面しているため、地元で獲れた新鮮な魚介類を使った料理が豊富です。  地元の市場や魚料理専門店で、その日の旬の魚を使った料理を尋ねてみましょう。  刺身、寿司、煮付けなど、様々な調理法で楽しめます。


3. **豊後料理に含まれる魚料理:** テキストに「豊後料理」が登場します。これは大分の郷土料理をアレンジした料理で、地元の食材をふんだんに使用している可能性が高いです。豊後料理を提供するレストランを探し、メニューに含まれる魚料理を試してみるのも良いでしょう。


これらの提案は、テキストの情報と一般的な知識に基づいた推測です。  より具体的な情報を得るには、大分市の観光情報サイトやグルメサイトなどを参照することをお勧めします。

FunctionCallingの実装②

ここから前のセクション同様FunctionCallingの実装にはいります。

functions = {
    "get_weather": get_spa,
    "get_gourmets":get_food,
}

model = genai.GenerativeModel(
    model_name="gemini-1.5-flash-latest",
    tools=functions.values(),
)

prompt = "あなたは今、佐伯市の魚を使った3つのおいしい新鮮な食べ物を紹介してください。また、おすすめの温泉も紹介してほしいです。"
response = model.generate_content(prompt)
response.candidates[0].content.parts
print(response.candidates[0].content.parts)
time.sleep(4)

• 出力

[text: "佐伯市には新鮮な魚介が豊富なので、おいしい食べ物がたくさんありますね！3つのオススメと温泉を以下にご紹介します。
\n\n**佐伯市で食べたい新鮮な魚料理3選**\n\n1. **関アジ・関サバの刺身:** 佐伯市沖で獲れる関アジと関サバは、その新鮮さと脂の乗りが絶品です。新鮮な刺身でいただくのが一番！プリプリとした食感と、上品な旨みが口の中に広がります。色々な部位を食べ比べてみるのも良いでしょう。
\n\n2. **豊後水道産の魚介を使った海鮮丼:**  佐伯市は豊後水道に面しており、様々な種類の魚介が獲れます。新鮮な魚介をたっぷり使った海鮮丼は、その日の漁によって内容が変わるのも楽しみの一つ。色々な魚介の味が一度に楽しめる贅沢な一品です。
\n\n3. **カボスを使った魚料理:** 佐伯市はカボスが特産品。爽やかな酸味が魚介類の旨みを引き立てます。カボスを使った焼き魚や煮付け、あるいは、新鮮な刺身にカボスを絞っていただくのもおすすめです。柑橘の香りが食欲をそそります。
\n\n\n**おすすめの温泉**\n\n佐伯市周辺には、多くの温泉がありますので、いくつか候補を挙げて、好みで選んでみて下さい。具体的な温泉名は、もう少し情報が必要です。例えば、「海の近くの温泉が良い」とか、「日帰り温泉が良い」など、希望があれば、より具体的な温泉名を紹介できます。
\n\n\nより具体的な情報が欲しい場合は、どのようなタイプの温泉がお好みかなど、もう少し詳細なご要望をお聞かせください。\n"
]

実はここまでしかうまく実装することが出来ていません。
プロンプトからモデルがどの関数を参照するべきかがわかっていないためです。

この先の実装方法や工夫の仕方がわかる方は教えていただけますと幸いです。

終わりに

何ともしまりの悪い記事となってしまいました笑

天気APIの実装の方は参考にしたサイトが優秀なこともあり、何とか実装できましたのでそちらを参考にしてください！

また、有識者の方々、私にfunctioncallingの実装方法を教えていただきたいです！
コメントお待ちしております！