【Azure AI Agent Service】- Tracing機能を試す

執筆日

2024/12/31

やること

Azure AI Agent Serviceには、Tracing機能がある。

Tracing機能とは、特定のエージェント実行に関係する各プリミティブの入力と出力を、それが呼び出された順序で明確に確認できる。

https://learn.microsoft.com/ja-jp/azure/ai-services/agents/concepts/tracing

ちょっとよくわからないので触ってみる

参考資料

https://github.com/Azure/azure-sdk-for-python/blob/main/sdk/ai/azure-ai-projects/samples/agents/sample_agents_basics_with_azure_monitor_tracing.py

手順

1. Azure AI Foundryを開く
2. 評価と改善>トレース　をクリックする
   
3. 新規作成 をクリックする
   
4. 名前を入力し、作成 をクリックする
   
5. Application inshgtsが作成されたことを確認する
6. ライブラリーをinstall

pip install opentelemetry-api azure-monitor-opentelemetry azure-identity azure-ai

7. 以下のコードを実行

main.py

import os, time
from azure.ai.projects import AIProjectClient
from azure.identity import DefaultAzureCredential

project_client = AIProjectClient.from_connection_string(
    credential=DefaultAzureCredential(),
    conn_str=os.environ["PROJECT_CONNECTION_STRING"],
)

# [START enable_tracing]
from opentelemetry import trace
from azure.monitor.opentelemetry import configure_azure_monitor

# Enable Azure Monitor tracing
application_insights_connection_string = project_client.telemetry.get_connection_string()
if not application_insights_connection_string:
    print("Application Insights was not enabled for this project.")
    print("Enable it via the 'Tracing' tab in your AI Foundry project page.")
    exit()
configure_azure_monitor(connection_string=application_insights_connection_string)

scenario = os.path.basename(__file__)
tracer = trace.get_tracer(__name__)

with tracer.start_as_current_span(scenario):
    with project_client:

        # [END enable_tracing]
        agent = project_client.agents.create_agent(
            model=os.environ["MODEL_DEPLOYMENT_NAME"], name="my-assistant", instructions="You are helpful assistant"
        )
        print(f"Created agent, agent ID: {agent.id}")

        thread = project_client.agents.create_thread()
        print(f"Created thread, thread ID: {thread.id}")

        message = project_client.agents.create_message(
            thread_id=thread.id, role="user", content="Hello, tell me a joke"
        )
        print(f"Created message, message ID: {message.id}")

        run = project_client.agents.create_run(thread_id=thread.id, assistant_id=agent.id)

        # Poll the run as long as run status is queued or in progress
        while run.status in ["queued", "in_progress", "requires_action"]:
            # Wait for a second
            time.sleep(1)
            run = project_client.agents.get_run(thread_id=thread.id, run_id=run.id)

            print(f"Run status: {run.status}")

        project_client.agents.delete_agent(agent.id)
        print("Deleted agent")

        messages = project_client.agents.list_messages(thread_id=thread.id)
        print(f"messages: {messages}")

8. トレースを確認する
   
   

9. 以下のブログの内容にTracing機能を追加して実行

https://zenn.dev/headwaters/articles/5c6edb44e40147

main.py

import os
import time
from azure.ai.projects import AIProjectClient
from azure.identity import DefaultAzureCredential
from azure.ai.projects.models import BingGroundingTool
from opentelemetry import trace
from azure.monitor.opentelemetry import configure_azure_monitor

# 初期設定
project_client = AIProjectClient.from_connection_string(
    credential=DefaultAzureCredential(),
    conn_str=<"プロジェクトの接続文字列">,
)

# [START enable_tracing] トレーシングを有効化
# Azure Monitor のトレーシングを有効にする
application_insights_connection_string = (
    project_client.telemetry.get_connection_string()
)
if not application_insights_connection_string:
    print("Application Insights was not enabled for this project.")
    print("Enable it via the 'Tracing' tab in your AI Foundry project page.")
    exit()

# OpenTelemetry を Azure Monitor に接続
configure_azure_monitor(connection_string=application_insights_connection_string)

# トレーサーを初期化
scenario = os.path.basename(__file__)
tracer = trace.get_tracer(__name__)
# [END enable_tracing]

# Bing Grounding Tool の設定
bing_connection = project_client.connections.get(connection_name=<"Bingのリソース名">)
conn_id = bing_connection.id
print(f"Bing Connection ID: {conn_id}")

# Bing Grounding Tool を初期化
bing = BingGroundingTool(connection_id=conn_id)

# トレーシングを開始し、エージェントを作成
with tracer.start_as_current_span(scenario):  # トレーシングスパンの開始
    with project_client:
        # エージェントの作成
        agent = project_client.agents.create_agent(
            model=<"model">,  # 使用するモデル
            name="my-assistant",  # エージェント名
            instructions="You are a helpful assistant",  # エージェントの指示
            tools=bing.definitions,  # Bing Tool を統合
            headers={"x-ms-enable-preview": "true"},
        )
        print(f"Created agent, ID: {agent.id}")

        # スレッドの作成
        thread = project_client.agents.create_thread()
        print(f"Created thread, ID: {thread.id}")

        # メッセージの作成
        message = project_client.agents.create_message(
            thread_id=thread.id,
            role="user",
            content="東京の天気について教えて",
        )
        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}")

        # 完了後、エージェントを削除
        project_client.agents.delete_agent(agent.id)
        print("Deleted agent")

        # メッセージ履歴を取得してログに記録
        messages = project_client.agents.list_messages(thread_id=thread.id)
        print(f"Messages: {messages}")

まとめ

Tracing機能を試してみました。
エージェントの各ステップを可視化できる機能でした。
各ステップごとに実行内容,応答時間が確認できるので、ボトルネックの特定と改善ができるのは便利だなと。