Microsoft Ignite 2024 で発表された Azure AI Agent Service SDK 版のパブリックプレビューが開始されました。
Azure AI Agent Service は Assistants API を基盤として Microsoft のマネージドサービスと組み合わせてエンタープライズ向けに機能拡張したサービスといえます。Bing Search、Microsoft Fabric、SharePoint、Azure AI Search、Azure Blob Storage、ローカル ファイル、さらには独自のライセンスデータからのデータを使用して、データをグラウンディングできるようになっています。
現在、Azure AI Agent Service の技術スタックは以下のようになっております。みなさんお気付きでしょう。そう、すでに Azure OpenAI の Assistants API によるエージェント開発メモ①で仕組みを勉強済みの方は習得した知識をそのまま利用できるということです。しかもさらに省コード化が図れるとのことで期待が高まります。
Assistants API との比較
- 柔軟なモデル選択: Azure OpenAI モデル（o1-preview, o1-mini, gpt-4o, gpt-4o-mini, gpt-4, 35-turbo）、または Llama 3.1-70B-instruct、Mistral-large-2407、Cohere command R+ が利用可能です
- 広範なデータ統合: Microsoft Bing、Microsoft SharePoint、Microsoft Fabric、Azure AI Search、その他の API など
- エンタープライズ グレードのセキュリティ: 安全なデータ処理、キーレス認証、パブリック エグレスなしでデータのプライバシーとコンプライアンスを確保
- ストレージ ソリューション: ストレージ リソースを完全に可視化して制御するために独自の Azure Blob ストレージを使用するか、安全で使いやすいプラットフォーム管理ストレージを使用可能
Ignite 2024 における将来的なビジョン
デモシナリオ
私の解説では定番のトラベルアシスタントシナリオで実装していきます。ナレッジツールとして Grounding with Bing Search と Azure AI Search を追加します。
事前準備
Basic エージェントおよび Standard エージェントの設定で自動デプロイボタンから構築するのが便利です。Standard エージェント設定では AI Hub、AI Project、Storage Account、Key Vault、Azure AI Search、AI Services リソースが自動デプロイされるのでこれを使用します。
プロジェクトに複数の Azure OpenAI リソースを接続するとエラーが出ました。AI Agent を利用するプロジェクトでは単一の Azure OpenAI リソースにする必要があります。
ライブラリ
azure-ai-projects は 12/14 時点で v1.0.0b3 です。
pip install azure-ai-projects
pip install azure-identity
接続設定
Azure AI Foundry の プロジェクト概要ページからプロジェクト接続文字列をコピーします。プロジェクトに接続しさえすればその下に紐づく Azure OpenAI モデルや Azure AI Search 等接続済みのリソースを利用できます。
import os
from azure.ai.projects import AIProjectClient
from azure.identity import DefaultAzureCredential
os.environ["PROJECT_CONNECTION_STRING"] = ""
project_client = AIProjectClient.from_connection_string(
credential=DefaultAzureCredential(), conn_str=os.environ["PROJECT_CONNECTION_STRING"]
)
1. エージェントの作成
1.1. 関数の定義
Function calling で実行される関数を定義しています。AI Agent では関数の定義と説明だけでなく、実装を含めることができます。ポイントは docstring を reStructuredText スタイルで記載しておくという点です。こうすることで、docstring がそのままスキーマ定義に反映され、これまで Assistants API で定義したような JSON 形式でスキーマをインプットする必要がなくなります。
import json
import datetime
from typing import Any, Callable, Set, Dict, List, Optional
import requests
# These are the user-defined functions that can be called by the agent.
#https://webservice.rakuten.co.jp/documentation/vacant-hotel-search
def search_vacant_hotels(
latitude: float,
longitude: float,
searchRadius: float,
checkinDate: str,
checkoutDate: str,
maxCharge: int = 50000,
adultNum: int = 1,
page: int = 1,
hits: int = 2,
response_format: str = 'json'
) -> str:
"""
楽天トラベルAPIを使用して空室のホテルを検索します。
:param latitude: (float) ホテル検索場所の緯度 (WGS84)。例: 35.6065914。
:param longitude: (float) ホテル検索場所の経度 (WGS84)。例: 139.7513225。
:param searchRadius: (float) 緯度経度検索時の検索半径 (単位: km)。0.1 から 3.0 まで指定可能。
:param checkinDate: (str) チェックイン日。形式: "yyyy-MM-dd"。年の指定がない場合は 2024年をデフォルトとします。
:param checkoutDate: (str) チェックアウト日。形式: "yyyy-MM-dd"。年の指定がない場合は 2024年をデフォルトとします。
:param maxCharge: (int) 上限金額 (単位: 円)。デフォルトは 50000 円。指定可能範囲: 0 から 999999999。
:param adultNum: (int) 宿泊者数。デフォルトは 1 人。指定可能範囲: 1 から 99。
:param page: (int) 検索結果のページ番号。デフォルトは 1。
:param hits: (int) 1ページあたりの取得件数。デフォルトは 2 件。
:param response_format: (str) APIレスポンスの形式を指定します。デフォルトは 'json'。
:return: APIレスポンスを辞書形式（デフォルトはJSON）で返します。
:rtype: str
"""
base_url = "https://app.rakuten.co.jp/services/api/Travel/VacantHotelSearch/20170426"
applicationId = "Your API key" # APIキーを設定
params = {
"applicationId": applicationId,
"format": response_format,
"page": page,
"hits": hits,
"latitude": latitude, # ex:35.6065914
"longitude": longitude, # ex:139.7513225
"searchRadius": searchRadius, #緯度経度検索時の検索半径(単位km), 0.1 to 3.0
"datumType": 1, # WGS84
"checkinDate": checkinDate, # yyyy-MM-dd
"checkoutDate": checkoutDate, # yyyy-MM-dd
"maxCharge": maxCharge, # 上限金額, int 0 to 999999999
"adultNum": adultNum # 宿泊者数, int 1 to 99
}
response = requests.get(base_url, params=params)
# レスポンス形式に応じて結果を処理
if response_format == 'json':
response_data = response.json()
if 'error' in response_data:
error_description = response_data.get('error_description', 'No error description provided.')
error = response_data.get('error', 'Unknown error')
# エラーメッセージを処理（またはログに記録）
print(f"Error: {error}, Description: {error_description}")
# エラー情報を含むレスポンスオブジェクトまたはメッセージを返す
return {"error": error, "error_description": error_description}
else:
# エラーがない場合は、通常通りレスポンスデータを返す
hotels = []
for hotel_group in response_data["hotels"]:
h = hotel_group["hotel"][0]["hotelBasicInfo"]
rooms = []
for item in hotel_group["hotel"][1:]:
if "roomInfo" in item:
roomplans = {"roomName": item["roomInfo"][0]["roomBasicInfo"]["roomName"],
"stayDate": item["roomInfo"][1]["dailyCharge"]["stayDate"],
"rakutenCharge": item["roomInfo"][1]["dailyCharge"]["rakutenCharge"]}
rooms.append(roomplans)
hotel = {"hotelNo": h["hotelNo"],
"hotelName": h["hotelName"],
"hotelSpecial": h["hotelSpecial"],
"access": h["access"],
"reviewAverage": h["reviewAverage"],
"hotelInformationUrl": h["hotelInformationUrl"],
"roomplans": rooms}
hotels.append(hotel)
return json.dumps(hotels, ensure_ascii=False)
else:
return response.text # XML形式の場合は、レスポンスのテキストをそのまま返す
def search_hotpepper_shops(
keyword: Optional[str] = None,
private_room: int = 0,
start: int = 1,
count: int = 3,
response_format: str = 'json'
) -> str:
"""
ホットペッパーグルメAPIを利用して飲食店を検索します。
:param keyword: (Optional[str]) 飲食店を検索するためのキーワード。店名、住所、駅名、お店ジャンルなどを指定できます。例: "大阪駅 和食"
:param private_room: (int) 個室ありの店舗で絞り込むかを指定します。0: 絞り込まない（デフォルト）, 1: 個室ありの店舗のみを検索。
:param start: (int) 検索結果の開始位置を指定します。デフォルトは1。
:param count: (int) 取得する結果数を指定します。デフォルトは3件。
:param response_format: (str) APIレスポンスの形式を指定します。デフォルトは'json'。
:return: APIレスポンスを指定形式（デフォルトはJSON）で返します。
:rtype: str
"""
base_url = "http://webservice.recruit.co.jp/hotpepper/gourmet/v1/"
api_key = "Your API Key" # APIキーを設定
params = {
"key": api_key,
"format": response_format,
"start": start,
"count": count,
"private_room": private_room, # 個室ありの店舗のみを検索, 0:絞り込まない, 1:絞り込む
#"budget": "B005"
}
# キーワードが指定されている場合、パラメータに追加
if keyword:
params["keyword"] = keyword
response = requests.get(base_url, params=params)
# レスポンス形式に応じて結果を処理
if response_format == 'json':
return response.json() # JSON形式のレスポンスを返す
else:
return response.text # XML形式の場合は、レスポンスのテキストをそのまま返す
# Statically defined user functions for fast reference
user_functions: Set[Callable[..., Any]] = {
search_vacant_hotels,
search_hotpepper_shops
}
最後にユーザー定義関数を user_functions に詰めておきます。
1.2. ツールの追加
ツールには toolset と tools が存在します。新しい toolset が非常に魅力的です。ツール呼び出しを解析したり、自分でツールを起動したり、応答を処理したりする必要がなくなります。
-
toolset:create_agentに関数の定義と説明だけでなく、その実装も提供。create_and_run_processメソッドで関数も実行できる -
tools:create_agentに関数の定義と説明のみを提供。requires_actionステータスを受け取ったら自分で実行結果を提供する必要がある。→これまでの Assistants API 同様
以下のように user_functions を FunctionTool に渡します。Code Interpreter を有効化するには、CodeInterpreterTool を定義します。
from azure.ai.projects import AIProjectClient
from azure.ai.projects.models import (
FunctionTool,
RequiredFunctionToolCall,
SubmitToolOutputsAction,
ToolOutput,
ToolSet,
CodeInterpreterTool,
AzureAISearchTool,
BingGroundingTool
)
functions = FunctionTool(user_functions)
code_interpreter = CodeInterpreterTool()
toolset = ToolSet()
toolset.add(functions)
toolset.add(code_interpreter)
1.3. エージェントの作成
パラメータの指定の仕方は Assistants API に似ていますね。モデルに gpt-4o-mini を指定します。
agent = project_client.agents.create_agent(
model="gpt-4o-mini",
name="Travel Assistant",
instructions="""
あなたは Contoso 社の社員の出張を支援するためのアシスタントです。あなたは以下の業務を遂行します。
- 旅程を作成します
- ホテルを検索したり予約します
- 交通機関を検索します
- 出張で行くべきレストランや居酒屋を提案します
- 出張にかかる概算費用を計算します
#制約事項
- ユーザーからのメッセージは日本語で入力されます
- ユーザーからのメッセージから忠実に情報を抽出し、それに基づいて応答を生成します。
- ユーザーからのメッセージに勝手に情報を追加したり、不要な改行文字 \n を追加してはいけません
""",
toolset=toolset,
headers={"x-ms-enable-preview": "true"},
)
print(f"Created agent, agent ID: {agent.id}")
agent.as_dict()
作成したエージェントオブジェクトを見てみると、tools の下に関数定義が格納されていることがわかります。
{'id': 'asst_l0UBDIefKlRvvoIiQaLh6sFv',
'object': 'assistant',
'created_at': 1734217458,
'name': 'Travel Assistant',
'description': None,
'model': 'gpt-4o-mini',
'instructions': '\nあなたは Contoso 社の社員の出張を支援するためのアシスタントです。あなたは以下の業務を遂行します。\n - 旅程を作成します\n - ホテルを検索したり予約します\n - 交通機関を検索します\n - 出張で行くべきレストランや居酒屋を提案します\n - 出張にかかる概算費用を計算します\n\n#制約事項\n - ユーザーからのメッセージは日本語で入力されます\n - ユーザーからのメッセージから忠実に情報を抽出し、それに基づいて応答を生成します。\n - ユーザーからのメッセージに勝手に情報を追加したり、不要な改行文字 \n を追加してはいけません\n\n',
'tools': [{'type': 'function',
'function': {'name': 'search_hotpepper_shops',
'description': 'ホットペッパーグルメAPIを利用して飲食店を検索します。',
'parameters': {'type': 'object',
'properties': {'keyword': {'type': ['string', 'null'],
'description': '(Optional[str]) 飲食店を検索するためのキーワード。店名、住所、駅名、お店ジャンルなどを指定できます。例: "大阪駅 和食"'},
...
1.4. エージェントを一覧表示
作成済みエージェントを一覧表示します。
agents = project_client.agents.list_agents()
for agt in agents.data:
print(f"{agt.id}: {agt.name}")
1.5. エージェントを取得
特定のエージェントにアクセスするには assistant_id を指定することで Agent オブジェクトを取得できます。
agent = project_client.agents.get_agent("Your-Assistant-ID")
agent.as_dict()
2. スレッドを作成
# Create a thread
thread = project_client.agents.create_thread()
thread.as_dict()
2.1. スレッドの取得
thread = project_client.agents.get_thread(thread_id="Your-Thread-ID")
thread.as_dict()
2.2. スレッドメッセージの一覧表示
messages = project_client.agents.list_messages(thread_id=thread.id)
messages.as_dict()
3. 同期的一括関数実行
create_and_process_run メソッドは Run の起動、ステータスポーリング、tool_calls オブジェクトの取得、関数実行、submit_tool_outputs_to_run メソッドの実行を一括で実行する新しいメソッドです。
関数ツールは、create_agent 時に toolset として提供された場合のみ呼び出されることに注意してください。
message = project_client.agents.create_message(
thread_id=thread.id,
role="user",
content="2/8に有楽町駅で1人、イタリアンの美味しい店探して",
)
print(f"Created message, ID: {message.id}")
run = project_client.agents.create_and_process_run(thread_id=thread.id, assistant_id=agent.id)
print(f"Run finished with status: {run.status}")
if run.status == "failed":
print(f"Run failed: {run.last_error}")
これは便利ですね！Assistants API の時は各処理を別々に実行していました。
参考 Run ステータス遷移動画解説
4. メッセージの確認
Run が完了したので、スレッドのメッセージを一覧表示して、エージェントが追加したメッセージを確認できます。メッセージは逆順に並ぶので以下ではリバースして表示しています。
from azure.ai.projects.models import MessageTextContent
messages = project_client.agents.list_messages(thread_id=thread.id)
# The messages are following in the reverse order,
# we will iterate them and output only text contents.
for data_point in reversed(messages.data):
last_message_content = data_point.content[-1]
if isinstance(last_message_content, MessageTextContent):
print(f"{data_point.role}: {last_message_content.text.value}")
user:2/8に有楽町駅で1人、イタリアンの美味しい店探して
assistant: 有楽町駅周辺での美味しいレストランを3軒ご紹介します。
1. **トラットリア・AI 有楽町店**
- **住所**: 東京都千代田区有楽町２－４－１ 有楽町プラザ３Ｆ
- **アクセス**: JR有楽町駅から徒歩2分／地下鉄日比谷駅A3出口から徒歩3分
- **ジャンル**: 本格イタリアンとワインが楽しめるトラットリア
- **予算**: 2500円（通常平均）／1200円（ランチ平均）
- **営業時間**: 月～日、祝日、祝前日: 11:00～23:00 （料理L.O. 22:00 ドリンクL.O. 22:30）
- **定休日**: 無
2. **グリル&バー サンシャイン 有楽町**
- **住所**: 東京都千代田区有楽町２－８－４ サンシャインビル５Ｆ
- **アクセス**: JR有楽町駅徒歩3分／地下鉄銀座駅徒歩5分
- **ジャンル**: ステーキとワインが楽しめる
- **予算**: 4500円（通常平均）／1000円（ランチ平均）
- **営業時間**: 月～日、祝日、祝前日: 11:00～23:00 （料理L.O. 22:00 ドリンクL.O. 22:30）
- **定休日**: 不定休
3. **スシバー 銀座 有楽町店**
- **住所**: 東京都千代田区有楽町２－２－３ 銀座ビル１Ｆ
- **アクセス**: JR有楽町駅徒歩1分／地下鉄銀座駅徒歩2分
- **ジャンル**: 新鮮なネタを使った寿司
- **予算**: 3000円（通常平均）／1200円（ランチ平均）
- **営業時間**: 月～日、祝日、祝前日: 11:00～22:00 （料理L.O. 21:30 ドリンクL.O. 21:45）
- **定休日**: 無
これらのレストランはそれぞれ異なるジャンルのお料理を提供しており、お好みに合わせて選べます。
（オプション）各プロセスを別々に実行
Run の起動、ステータスポーリング、tool_calls オブジェクトの取得、関数実行、submit_tool_outputs_to_run メソッドの実行を別々に行うこともできます。一応以下にまとめたコードを載せておきます。
import time
message = project_client.agents.create_message(
thread_id=thread.id,
role="user",
content="2/14に東京近辺で1人で泊まれるビジネスホテルを探しています。",
)
print(f"Created message, ID: {message.id}")
run = project_client.agents.create_run(thread_id=thread.id, assistant_id=agent.id)
print(f"Created run, ID: {run.id}")
while run.status in ["queued", "in_progress", "requires_action"]:
time.sleep(1)
run = project_client.agents.get_run(thread_id=thread.id, run_id=run.id)
if run.status == "requires_action" and isinstance(run.required_action, SubmitToolOutputsAction):
tool_calls = run.required_action.submit_tool_outputs.tool_calls
if not tool_calls:
print("No tool calls provided - cancelling run")
project_client.agents.cancel_run(thread_id=thread.id, run_id=run.id)
break
tool_outputs = []
for tool_call in tool_calls:
if isinstance(tool_call, RequiredFunctionToolCall):
try:
print(f"Executing tool call: {tool_call}")
output = functions.execute(tool_call)
response = {"answer": output, "success": True}
tool_outputs.append(
ToolOutput(
tool_call_id=tool_call.id,
output=json.dumps(response, ensure_ascii=False),
)
)
except Exception as e:
print(f"Error executing tool_call {tool_call.id}: {e}")
print(f"Tool outputs: {tool_outputs}")
if tool_outputs:
project_client.agents.submit_tool_outputs_to_run(
thread_id=thread.id, run_id=run.id, tool_outputs=tool_outputs
)
print(f"Current run status: {run.status}")
print(f"Run completed with status: {run.status}")
5. Bing 検索ツール
Azure Portal から新たに Grounding with Bing Search リソースを作成します。この Grounding with Bing Search リソースは、Azure AI Agent、AI Project、その他のリソースと同じリソース グループに作成してください。
5.1. 接続設定
Azure AI Foundry の管理センターから接続設定をカスタム API キーとして追加します。
接続設定が完了したら、以下のように接続設定をロードして BingGroundingTool を作成します。
bing_connection = project_client.connections.get(
connection_name="bing-grounding-hana1"
)
conn_id = bing_connection.id
print(conn_id)
# Initialize agent bing tool and add the connection id
bing = BingGroundingTool(connection_id=conn_id)
toolset.add(bing)
注意
Grounding with Bing Search を使用するには gpt-35-turbo-0125, gpt-4-0125-preview, gpt-4-turbo-2024-04-09, gpt-4o-2024-05-13, gpt-4o-2024-08-06 のみがサポートされています。
5.2. 実行結果
user: 最新の京都のイベントニュースを教えてください
assistant: 最近の京都のイベントニュースには、以下のようなものがあります：
1. 東映太秦映画村で「ゆく年くる年えいがむら 2024→2025」が12月21日から...【1†source】。
2. 京都市左京区の関西日仏学館でフランス伝統のクリスマス・マーケットが開催中です【5†source】。
3. 京都鉄博では12月の土曜・日曜日にイベントが開催され、100系新幹線の展示...
5.3. 検索パラメータの取得
検索パラメータは Run Step から見つけられるが、文字化けしているのでデコードする。source ページの URL はどこから取る？
run_steps = project_client.agents.list_run_steps(run_id=run.id, thread_id=thread.id)
run_steps_data = run_steps['data']
print(f"Last run step detail: {run_steps_data}")
import urllib.parse
encoded_text = run_steps_data[1]["step_details"]["tool_calls"][0]["bing_grounding"]["requesturl"]
decoded_text = bytes(encoded_text, 'latin1').decode('utf-8')
final_text = urllib.parse.unquote(decoded_text)
print(final_text)
https://api.bing.microsoft.com/v7.0/search?q="最新 京都 イベント ニュース", recency_days=30)"
ご利用には Grounding with Bing Search の利用規約および法的要件を参照ください。
6. Azure AI Search
既存 Azure AI Search のインデックスを検索対象にすることができます。
ai_search_connection = project_client.connections.get(
connection_name="hub-demo-f5yj-connection-AISearch"
)
conn_id = ai_search_connection.id
print(conn_id)
# Initialize agent AI search tool and add the search index connection id
ai_search = AzureAISearchTool(index_connection_id=conn_id, index_name="ragtest1-index")
#toolset.add(ai_search)
agent = project_client.agents.create_agent(
model="gpt-4o-mini",
name="my-assistant",
instructions="You are a helpful assistant",
tools=ai_search.definitions,
tool_resources=ai_search.resources,
headers={"x-ms-enable-preview": "true"},
)
ただし、うまく取ってこれないケースもあり、内部の Grounding がどうなっているのか可視化する必要があります。本機能は引き続きウォッチします。
ただ Function calling でクライアントごと実装することもできるので、そちらの方が断然自由度が高いですね。
7. 付録 AI Agent 操作でよく使う API
Run のキャンセル
in_progress 状態の Run をキャンセルします。
project_client.agents.cancel_run(thread_id=thread.id, run_id=run.id)
7.1. スレッドの削除
project_client.agents.delete_thread("Your-Thread-ID")
7.2. エージェントの削除
project_client.agents.delete_agent("Your-Assistant-ID")
8. 実行トレース
AI Agent API および Function calling の各実行結果を Azure AI Foundry のトレースに登録できます。Application Insights との連携が必要です。
9. データの保存場所
Azure AI Agent Service はステートフルなサービスなので以下のようにデータが保存されます。
- Azure OpenAI リソースと同じ GEO 内の顧客の Azure テナントで Azure AI Agent Service を構成するときに作成される Azure OpenAI リソースに保存される
- 保存時に二重で暗号化可能。既定では Microsoft の AES-256 暗号化を使用し、オプションで顧客管理キーを使用 (ただし、プレビュー機能では顧客管理キーがサポートされない場合がある)。
- 顧客はいつでも削除できる
GitHub
もっと細かく一連の動作を確認するにはこちらの Notebook を参照ください。
参考