はじめに
前回に引き続き,Ciscoのクラウド管理型ネットワーキングソリューションである「Cisco Meraki」が提供するAPIを使って,自社のデータも取得可能なAIチャットボットを簡易的に実装した様子をまとめていきます.
前回は,チャットボットを実装する際のアーキテクチャや,その際に使用するフレームワークであるFunction Callingの概要について触れました.
今回は,Function Callingを使って実際に処理を書いていく際のポイントや注意点についてシェアできればと思います.
Function Calling実装時のポイント
実際にPythonコードとして実装する際には,LLMの性質に関して押さえておくべき重要なポイントがあります.
LLMはその仕組み上,自らAPIを呼び出すことはできません.APIの呼び出し処理は,あくまでチャットボット本体となるPythonスクリプト内に記述する必要があります.
つまり,LLMは「APIを使いたい」という意図は理解できますが,その実行自体はPython側で行う必要があるということです.その際のPython側の実際の処理としては,PythonスクリプトがAPIサーバーへHTTPリクエストを送り,レスポンスを受け取るという流れになります.
通常,HTTPリクエストには,取得したい情報に応じたパラメータを指定します.Function Callingを使うと,LLMはユーザーのプロンプトの内容を解釈し,PythonスクリプトがAPI呼び出し時に含めるべきパラメータを提示してくれます.
ただし,ここで注意すべき点があります.多くのAPIは継続的にアップデートされ,URLや受け付けるパラメータが変化する可能性があります.したがって,LLMが返す情報が常に正確とは限りません.そのため,もし利用するAPIが明確に決まっている場合は,そのAPIの仕様（取得できる情報の内容や必要なパラメータ）をあらかじめLLMにプロンプトとして与えておくことを推奨します.
Function Calling活用時のAPI処理のイメージとしては以下のようになります.
（この説明では分かりやすさのため,架空の「最新ニュース取得API」と「天気取得API」を例に記載）
・LLMに必要な情報を事前にインプット
さらに,API利用時にはもう一つ注意すべきポイントがあります.
それは,「必要なパラメータが事前にわかっていなければ,LLMだけでは正しいAPIリクエストを組み立てられない」という点です.例えばMeraki APIでは,各Meraki機器が設置されているユーザー拠点を「networkId」というパラメータで管理しています.Python側で特定拠点の機器情報を取得するには,その拠点に対応するnetworkIdをパラメータとしてAPIリクエストに含める必要があります.
問題は,このnetworkIdがインターネット上で一般公開されている情報ではなく,各ユーザー拠点に固有の値だという点です.つまり,LLMは「どのユーザー拠点（例：シドニー）がどのnetworkIdに対応するか」という対応関係を知りません.
そのため,この対応表はPythonスクリプト内にあらかじめ定義しておき,プロンプトの一部としてLLMに渡しておく必要があります.こうしておけば,ユーザーが例えば「シドニー拠点の情報を知りたい」と入力したとき,LLMは「シドニー拠点＝networkId: xxxx_xxxx_xxxx」という対応関係を参照し,適切なパラメータを付けたAPIリクエストを組み立てられるようになります.
以上のポイントを踏まえつつ,実際に作成したLambda関数（Python）は以下になります.
Lambda関数（Python）
このスクリプトは,WebexからのWebhookを受け取り,Meraki APIやOpenAI APIと連携して応答を生成し,Webexにメッセージを返します.
1.事前準備（環境変数の参照,DBの初期設定等）
# 必要なライブラリをインポート
import json
import requests
import os
import boto3
from openai import OpenAI
# 環境変数からWebex Botの認証トークンを取得
WEBEX_TEAMS_BOT_TOKEN = os.environ['WEBEX_TEAMS_BOT_TOKEN']
# Webex APIのエンドポイントURLを定義
WEBEX_TEAMS_API_URL = "https://webexapis.com/v1"
# 環境変数からOpenAIのAPIキーを取得
openai_api_key = os.environ['OPENAI_API_KEY']
# 環境変数からMerakiのAPIキーを取得
meraki_api_key = os.environ['MERAKI_API_KEY']
# 環境変数からメッセージを投稿するWebexルームのIDを取得
ROOMID = os.environ['ROOMID']
# 環境変数からMerakiのOrganization IDを取得
organization_id = os.environ['ORGANIZATION_ID']
# DynamoDBに接続するための設定
dynamodb = boto3.resource('dynamodb', region_name='ap-northeast-1') # リージョンを適切に設定
# 使用するDynamoDBのテーブル名を定義
table_name = 'My_ChatHistory'
# 指定したテーブル名のテーブルオブジェクトを取得
table = dynamodb.Table(table_name)
2.拠点とネットワークIDの対応表の定義
networks = [
{
"organizationId": "組織IDを入力",
"name": "London",
"id": "xxxxxxx"
},
{
"organizationId": "組織IDを入力",
"name": "Sydney",
"id": "yyyyyyy"
},
# (以下、他のネットワーク情報が続く)
{
"organizationId": "組織IDを入力",
"name": "Dallas",
"id": "zzzzzzz"
},
]
3.DB操作の設定
# DynamoDBにチャット履歴を保存する関数
def save_chat_history(user_id, messages):
"""ユーザーのチャット履歴を保存"""
table.put_item(
Item={
'UserId': user_id, # パーティションキーとしてユーザーIDを設定
'Messages': messages # 保存するメッセージ履歴
}
)
# ・DynamoDBからチャット履歴を読み込む関数
def load_chat_history(user_id):
"""ユーザーのチャット履歴をロード"""
response = table.get_item(
Key={
'UserId': user_id
}
)
return response.get('Item', {}).get('Messages', [])
4.Function Callingで扱うAPIの設定
# Meraki APIを呼び出してライセンスの概要情報を取得する関数
def get_meraki_license_overview():
url = f"https://api.meraki.com/api/v1/organizations/{organization_id}/licenses/overview"
headers = {
"Authorization": f"Bearer {meraki_api_key}", # 認証情報
"Accept": "application/json" # レスポンス形式
}
response = requests.get(url, headers=headers)
response.raise_for_status()
return response.json()
# Meraki APIを呼び出して指定したネットワークのセンサー情報を取得する関数
def get_meraki_sensor_readings(network_id):
try:
url = f"https://api.meraki.com/api/v1/organizations/{organization_id}/sensor/readings/latest?networkId={network_id}"
headers = {
"Authorization": f"Bearer {meraki_api_key}",
"Accept": "application/json"
}
response = requests.get(url, headers=headers)
response.raise_for_status()
data = response.json()
return data
except requests.exceptions.RequestException as e:
raise
# Meraki APIを呼び出して指定したネットワークのアラート情報を取得する関数
def get_meraki_network_alerts(network_id):
url = f"https://api.meraki.com/api/v1/organizations/{organization_id}/assurance/alerts?networkId={network_id}"
headers = {
"Authorization": f"Bearer {meraki_api_key}",
"Accept": "application/json"
}
response = requests.get(url, headers=headers)
response.raise_for_status()
return response.json()
5.各拠点に対応するネットワークIDを返す関数
def get_network_info(network_id):
"""指定されたネットワークIDの詳細情報を返す"""
for network in networks:
if network['id'] == network_id:
return network
return None
6.ユーザーからのメッセージを処理する関数
def respond_to_message(user_id, message_text):
# OpenAI APIクライアントを初期化
client = OpenAI(api_key=openai_api_key)
# DynamoDBからこのユーザーの過去のチャット履歴をロード
messages = load_chat_history(user_id)
# 今回のユーザーからの新しいメッセージを履歴に追加
messages.append({"role": "user", "content": message_text})
# `networks`リストの情報を整形して,後でシステムメッセージに含める準備
network_info = []
for network in networks:
network_info.append(
f"- Name: {network['name']}\n"
f" ID: {network['id']}\n"
f" Organization ID: {network['organizationId']}"
)
# OpenAIモデルに与えるシステムメッセージ（役割や背景情報）を構築
system_message = "以下のネットワーク情報を必要に応じて適宜参照し,ユーザーの質問に答えてください：\n\n" + "\n\n".join(network_info) + "\n\n以下のネットワーク情報のうち,「ID: {network['id']}」の部分が「network_id」に対応します."
# システムメッセージを履歴に追加
messages.append({"role": "system", "content": system_message})
# OpenAIのFunction Calling機能で使用するツールのリストを定義
tools = [
{
"type": "function",
"function": {
"name": "get_My_meraki_network_alerts",
"description": "Get network alerts for a specific location using its network ID",
"parameters": {
"type": "object",
"properties": {
"network_id": {
"type": "string",
"description": "The ID of the network you want to get alerts for",
},
},
"required": ["network_id"],
},
},
},
{
"type": "function",
"function": {
"name": "get_My_meraki_license_information",
"description": "Get information for My Meraki license",
"parameters": {
"type": "object",
"properties": {},
},
},
},
{
"type": "function",
"function": {
"name": "get_My_meraki_sensor_information",
"description": "Get a list of real-time environment data collected by Meraki sensors on each network",
"parameters": {
"type": "object",
"properties": {
"network_id": {
"type": "string",
"description": "The ID of the network you want to get sensor information for",
},
},
"required": ["network_id"],
},
},
},
]
# OpenAIのChat Completion APIを呼び出す
response = client.chat.completions.create(
model="gpt-4o-mini", # 使用するモデルを指定
messages=messages, # これまでの会話履歴を渡す
tools=tools, # 使用可能なツールを渡す
)
# APIからの最初のレスポンスメッセージを取得
response_message = response.choices.message
# レスポンスメッセージを辞書形式に変換して履歴に追加
messages.append(response_message.to_dict())
# レスポンスにツール呼び出し（Function Calling）のリクエストが含まれていた場合
if response_message.tool_calls:
for tool_call in response_message.tool_calls:
function_name = tool_call.function.name
function_args = json.loads(tool_call.function.arguments)
if function_name == "get_My_meraki_network_alerts":
network_info_obj = get_network_info(function_args["network_id"])
alerts = get_meraki_network_alerts(function_args["network_id"])
response_text = f"Network: {network_info_obj['name']}\nAlerts: {json.dumps(alerts)}"
elif function_name == "get_My_meraki_license_information":
license_info = get_meraki_license_overview()
response_text = f"My Meraki License Overview: {json.dumps(license_info)}"
elif function_name == "get_My_meraki_sensor_information":
network_info_obj = get_network_info(function_args["network_id"])
sensor_readings = get_meraki_sensor_readings(function_args["network_id"])
response_text = f"Network: {network_info_obj['name']}\nSensor Readings: {json.dumps(sensor_readings)}"
messages.append(
{
"tool_call_id": tool_call.id,
"role": "tool",
"name": function_name,
"content": response_text
}
)
# ツール実行結果を含めたメッセージ履歴を元に,再度OpenAI APIを呼び出し,最終的な応答を生成
second_response = client.chat.completions.create(
model="gpt-4o-mini",
messages=messages,
)
# 2回目のAPIからのレスポンスメッセージを取得
second_response_message = second_response.choices.message
# 最終的な応答テキストを取得
final_response_text = second_response_message.content
# 更新されたチャット履歴をDynamoDBに保存
save_chat_history(user_id, messages)
# 生成された最終応答をWebexに投稿
post_webex_message(final_response_text)
return "OK"
7.Webexにメッセージを投稿する関数
def post_webex_message(text):
url = f"{WEBEX_TEAMS_API_URL}/messages"
headers = {
"Authorization": f"Bearer {WEBEX_TEAMS_BOT_TOKEN}", # Botの認証トークン
"Content-Type": "application/json"
}
payload = {
"roomId": ROOMID, # 投稿先のルームID
"markdown": text # 投稿するメッセージ内容（Markdown形式）
}
response = requests.post(url, json=payload, headers=headers)
response.raise_for_status()
8.AWS Lambdaのハンドラ関数（Lambda実行の起点）
def lambda_handler(event, context):
try:
# Webhookから送られてきたリクエストのボディ部分をJSONとしてパース
webhook_data = json.loads(event['body'])
if webhook_data['resource'] == 'messages' and webhook_data['event'] == 'created':
# 作成されたメッセージのIDを取得
message_id = webhook_data['data']['id']
動作確認
Webex上の任意の部屋に事前設定が済んだ状態のBot (My-MerakiBot) を追加し,自社拠点のMerakiに関する様々な質問を投げかけてみます.
1.まずは挨拶から
どうやら準備万端のようです.
2.サンフランシスコ拠点のアラート情報を取得
サンフランシスコ拠点のアラートの情報（日時,概要,重要度等）を返してくれました.
回答に含まれている青色のリンクをクリックすることで,アラートの詳細ページにジャンプすることができました.
3.シドニー拠点の温湿度データを取得
「蒸し暑いかどうか」という問いに対し,適切な回答の生成には温湿度の情報が必要であることをLLMが自律的に判断してくれたようです.これにより,MerakiのIoTセンサーで収集した温湿度データをAPIで取得し,欲しかった回答を得ることができました.
まとめ
本記事では,Cisco Merakiが提供するAPIと,Open AIの文章生成用APIを組み合わせて,リアルタイムなネットワーク情報を取得できるAIチャットボットの実装例について,前回 (理論編)と今回 (実践編)の計二回に分けて紹介してきました.
従来のパターンマッチング方式に比べ,Function Callingを活用することで,より柔軟かつ直感的に必要なAPIを呼び出せる点が大きなメリットです.これにより,ユーザの意図に応じて適切なリアルタイムデータを取得し,より有用な回答を生成することが可能となります.
本記事が,AIチャットボットの開発やネットワーク運用の自動化に興味を持つ方々の参考になれば幸いです.