Lifematics Corporate Blog

Lifematics社のコーポレートブログへようこそ!

Microsoft Copilot Studio でLLMOpsを行うための第一歩!

Microsoft Copilot Studio でLLMOpsを行うための第一歩!

はじめに

Lifematicsのイノベーションチームの松本です。 前回のCopilot Studioに関する記事では、Copilot Studioを利用して社内情報を参照したチャットボットを構築する方法を紹介しました。 しかしチャットボットを始めとしたAIエージェントの運用においては、構築するだけではなく継続的な品質管理が重要です。Copilot StudioではGUIで簡単にAIエージェントの構築を行うことができるのが強みですが、2025年10月現在、回答精度の評価機能にはAzure AI Foundryと比較すると限界があり、特に以下のような課題に直面します:

  • 複数のテストケースを繰り返し実行する手間
  • 回答精度の定量的な評価の難しさ
  • モデルやプロンプト変更時の影響確認の煩雑さ

本記事では、Microsoft 365 Agents SDKを活用してコマンドラインからAIエージェントへのクエリを一括実行し、LLMOpsの実現に向けた基盤を構築する方法を紹介します。

対象読者

  • Microsoft Copilot Studioでエージェントを開発・運用している方
  • AIエージェントの品質管理やテスト自動化に興味がある方
  • LLMOpsの実践方法を探している方
  • コマンドラインの利用に抵抗がない方

前提条件

  • Microsoft Copilot Studioでエージェントが作成済み
  • Pythonの実行環境(3.9以上)
  • Gitコマンドが利用可能
  • Entra ID上でアプリケーションに対して管理者の同意を付与する権限

※本記事ではPythonを利用しますが、Microsoft 365 Agents SDKJavaScript/TypeScriptやC#でも利用可能です。

Microsoft 365 Agents SDKとは

Microsoft 365 Agents SDKは、Microsoft Copilot Studioで作成したエージェントとプログラマティックに対話するためのSDKです。このSDKを使用することで、GUIを経由せずにエージェントにメッセージを送信し、応答を受け取ることができます。

主な機能

  • エージェントへのメッセージ送信
  • 会話セッションの管理
  • JWT認証トークンによる安全な接続
  • Azure AD統合による権限管理

事前準備

プログラムを実行する前に、Copilot StudioのエージェントIDの取得と、Entra ID上でアプリケーション登録と必要なAPIアクセス許可の設定を行います。少々手順が長いですが、Entra IDの設定は環境ごとに一度行えば済みます。

1. Copilot StudioのエージェントIDの取得

  1. Copilot Studioにログインします。
  2. 対象のエージェントを選択します。
  3. エージェントの設定画面で 上級 => メタデータに移動し、以下の値(スクリーンショット内赤枠の部分)をメモしておきます。
    • 環境ID
    • テナントID
    • スキーマ
    • ※エージェントアプリIDは使用しません。

2. Entra IDでのアプリケーション登録とAPIアクセス許可の設定

  1. Azureポータルにログインします。
  2. Microsoft Entra ID に移動する
  3. Microsoft Entra ID で「管理」→「アプリの登録」→「新規登録」をクリックし、以下の情報を入力してアプリケーションを登録します。
    1. 名前を入力する(例: CopilotStudioClientApp)
    2. 「この組織ディレクトリのみに含まれるアカウント」を選択する
    3. リダイレクトURI の「プラットフォームを選択」リストで、「パブリッククライアント/ネイティブ(モバイルとデスクトップ)」を選択する
    4. その右の入力欄に http://localhost と入力する(注意:HTTPSではなくHTTPを使用
    5. 登録をクリックする
  4. 新しく作成されたアプリケーションで
    1. 概要ページで、「アプリケーション(クライアント)ID」をメモする。また、「ディレクトリ(テナント)ID」が事前に作成済みのエージェントのテナントIDと一致することを確認する。
    2. 「管理」 セクションの「APIのアクセス許可」に移動する
    3. 「アクセス許可の追加」をクリックする
      1. 表示されるサイドパネルで、所属する組織で使用しているAPI タブをクリックする
      2. Power Platform API を検索する
        1. Power Platform API が表示されない場合は、Power Platform APIをテナントに追加する必要があります。Power Platform API Authentication に移動し、ステップ2の指示に従ってPower Platform Admin APIをテナントに追加してください。
      3. 委任されたアクセス許可 リストで、CopilotStudio を選択し、CopilotStudio.Copilots.Invoke をチェックする
      4. アクセス許可の追加 をクリックする
    4. {組織名}に管理者の同意を与えます をクリックする

※↓は完了後のイメージです。

Pythonプログラムの実行

それではプログラムの実行に移ります。MicrosoftSDKを利用したサンプルプログラムを提供しているので、これをベースにします。

1. ソースコードの入手

まず、GitHubからサンプルコードをクローンします。

git clone git@github.com:microsoft/Agents.git
cd Agents/samples/python/copilotstudio-client/

2. 環境構築

仮想環境を作成したうえで必要なパッケージをインストールします。

python -m venv copilot-studio-venv
source copilot-studio-venv/bin/activate  # Windowsの場合は `copilot-studio-venv\Scripts\activate`
pip install -r requirements.txt

3. 環境変数の設定

環境変数を設定します。env.TEMPLATEをコピーして.envを作成します。

cp env.TEMPLATE .env

.envファイルを開き、事前準備でメモしておいた値を設定します。

COPILOTSTUDIOAGENT__ENVIRONMENTID=          # Copilot Studio上の環境ID
COPILOTSTUDIOAGENT__SCHEMANAME=             # Copilot Studio上のスキーマ名
COPILOTSTUDIOAGENT__TENANTID=               # テナントID
COPILOTSTUDIOAGENT__AGENTAPPID=             # Entra IDの「アプリケーション(クライアント)ID

4. サンプルプログラムの実行

これで(ようやく)サンプルプログラムの実行準備が整ったので、実行してみます。

python -m src.main

プログラムを実行すると、Webブラウザが起動してMicrosoftアカウントのログイン画面が表示されます。Copilot Studioのエージェントにアクセスできるアカウントでログインしてください。 ログインが完了すると、コマンドラインでエージェントの応答が表示されたあと、入力待ちになります。

Suggested Actions: 
こんにちは、セルルン です!何か質問がありますか?

>>>: 

※「セルルン」はサンプルでCopilot Studioで作成したエージェントの名前であり、弊社のマスコット名でもあります。

エージェントに対してクエリを入力すると、無事に応答が返ってきました! 🎉

>>>: こんにちは、いい天気ですね。

こんにちは!本当にいい天気ですね。今日は何か特別な予定がありますか?
AI-generated content may be incorrect

会話を終わりたいときは、'exit', もしくは'quit'を入力すると終了します。

5. 一括クエリ実行の実装

src/main.pyを編集して、一括クエリ実行の仕組みを追加します。以下のサンプルコードの内容に置き換えます。

# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License.

# enable logging for Microsoft Agents library
# for more information, see README.md for Quickstart Agent
import logging
ms_agents_logger = logging.getLogger("microsoft_agents")
ms_agents_logger.addHandler(logging.StreamHandler())
ms_agents_logger.setLevel(logging.INFO)

import sys
from os import environ
import asyncio
import webbrowser
import atexit
import csv
from pathlib import Path

from dotenv import load_dotenv

from msal import PublicClientApplication

from microsoft_agents.activity import ActivityTypes, load_configuration_from_env
from microsoft_agents.copilotstudio.client import (
    ConnectionSettings,
    CopilotClient,
)

from .local_token_cache import LocalTokenCache

logger = logging.getLogger(__name__)

load_dotenv()

TOKEN_CACHE = LocalTokenCache("./.local_token_cache.json")


def save_token_cache():
    """プログラム終了時にトークンキャッシュを保存する"""
    try:
        TOKEN_CACHE.serialize()
        logger.debug("Token cache saved on exit")
    except Exception as e:
        logger.exception("Error saving token cache on exit")


# プログラム終了時にトークンキャッシュを保存
atexit.register(save_token_cache)


async def open_browser(url: str):
    logger.debug(f"Opening browser at {url}")
    await asyncio.get_event_loop().run_in_executor(None, lambda: webbrowser.open(url))


def acquire_token(settings: ConnectionSettings, app_client_id, tenant_id):
    pca = PublicClientApplication(
        client_id=app_client_id,
        authority=f"https://login.microsoftonline.com/{tenant_id}",
        token_cache=TOKEN_CACHE,
    )

    token_request = {
        "scopes": ["https://api.powerplatform.com/.default"],
    }
    accounts = pca.get_accounts()
    retry_interactive = False
    token = None
    try:
        if accounts:
            response = pca.acquire_token_silent(
                token_request["scopes"], account=accounts[0]
            )
            token = response.get("access_token")
        else:
            retry_interactive = True
    except Exception as e:
        retry_interactive = True
        logger.error(
            f"Error acquiring token silently: {e}. Going to attempt interactive login."
        )

    if retry_interactive:
        logger.debug("Attempting interactive login...")
        response = pca.acquire_token_interactive(**token_request)
        token = response.get("access_token")

    # トークン取得後にキャッシュを明示的に保存
    TOKEN_CACHE.serialize()
    logger.debug("Token cache saved after token acquisition")
    
    return token


def create_client():
    settings = ConnectionSettings(
        environment_id=environ.get("COPILOTSTUDIOAGENT__ENVIRONMENTID"),
        agent_identifier=environ.get("COPILOTSTUDIOAGENT__SCHEMANAME"),
        cloud=None,
        copilot_agent_type=None,
        custom_power_platform_cloud=None,
    )
    token = acquire_token(
        settings,
        app_client_id=environ.get("COPILOTSTUDIOAGENT__AGENTAPPID"),
        tenant_id=environ.get("COPILOTSTUDIOAGENT__TENANTID"),
    )
    copilot_client = CopilotClient(settings, token)
    return copilot_client


async def ainput(string: str) -> str:
    await asyncio.get_event_loop().run_in_executor(
        None, lambda s=string: sys.stdout.write(s + " ")
    )
    return await asyncio.get_event_loop().run_in_executor(None, sys.stdin.readline)


async def ask_single_question(copilot_client, question):
    """単一の質問をCopilotに送信し、回答を取得する"""
    print(f"質問: {question}")
    
    answer_text = ""
    replies = copilot_client.ask_question(question)
    
    async for reply in replies:
        if reply.type == ActivityTypes.message:
            print(f"回答: {reply.text}")
            answer_text += reply.text + " "
            if reply.suggested_actions:
                for action in reply.suggested_actions.actions:
                    print(f" - {action.title}")
        elif reply.type == ActivityTypes.end_of_conversation:
            print("会話終了")
            break
    
    return answer_text.strip()


def read_csv_questions(csv_file_path):
    """CSVファイルから質問を読み取る"""
    questions = []
    try:
        with open(csv_file_path, 'r', encoding='utf-8-sig') as file:
            reader = csv.DictReader(file)
            for row in reader:
                if 'question' in row and row['question'].strip():
                    questions.append(row)
        print(f"{len(questions)}個の質問を読み取りました")
        return questions
    except FileNotFoundError:
        print(f"CSVファイルが見つかりません: {csv_file_path}")
        sys.exit(1)
    except Exception as e:
        print(f"CSVファイル読み取りエラー: {e}")
        sys.exit(1)


def save_csv_with_answers(csv_file_path, questions_with_answers):
    """回答を追加してCSVファイルを保存する"""
    if not questions_with_answers:
        return
    
    # 元のCSVのフィールド名を取得し、answerを追加
    fieldnames = list(questions_with_answers[0].keys())
    if 'answer' not in fieldnames:
        fieldnames.append('answer')
    
    try:
        with open(csv_file_path, 'w', newline='', encoding='utf-8-sig') as file:
            writer = csv.DictWriter(file, fieldnames=fieldnames)
            writer.writeheader()
            writer.writerows(questions_with_answers)
        print(f"結果を保存しました: {csv_file_path}")
    except Exception as e:
        print(f"CSV保存エラー: {e}")


async def ask_question(copilot_client, conversation_id):
    query = (await ainput("\n>>>: ")).lower().strip()
    if query in ["exit", "quit"]:
        print("Exiting...")
        return
    if query:
        replies = copilot_client.ask_question(query, conversation_id)
        async for reply in replies:
            if reply.type == ActivityTypes.message:
                print(f"\n{reply.text}")
                if reply.suggested_actions:
                    for action in reply.suggested_actions.actions:
                        print(f" - {action.title}")
            elif reply.type == ActivityTypes.end_of_conversation:
                print("\nEnd of conversation.")
                sys.exit(0)
        await ask_question(copilot_client, conversation_id)


async def process_csv_questions(csv_file_path):
    """CSVファイルから質問を読み取り、自動実行して結果を保存する"""
    print(f"CSVファイルを処理中: {csv_file_path}")
    
    # CSVから質問を読み取り
    questions = read_csv_questions(csv_file_path)
    if not questions:
        print("処理する質問がありません")
        return
    
    # Copilotクライアントを作成
    copilot_client = create_client()
    act = copilot_client.start_conversation(True)
    
    print("\nSuggested Actions: ")
    async for action in act:
        if action.text:
            print(action.text)
    
    # 各質問を処理
    questions_with_answers = []
    for i, question_row in enumerate(questions, 1):
        question = question_row['question']
        print(f"\n[{i}/{len(questions)}] 処理中...")
        
        try:
            answer = await ask_single_question(copilot_client, question)
            question_row['answer'] = answer
        except Exception as e:
            print(f"エラーが発生しました: {e}")
            question_row['answer'] = f"エラー: {str(e)}"
        
        questions_with_answers.append(question_row)
        
        # 進捗表示
        print(f"完了: {i}/{len(questions)}")
    
    # 結果をCSVに保存
    save_csv_with_answers(csv_file_path, questions_with_answers)
    print("\n全ての質問の処理が完了しました")


async def main_interactive():
    """インタラクティブモード(元の機能)"""
    copilot_client = create_client()
    act = copilot_client.start_conversation(True)
    print("\nSuggested Actions: ")
    async for action in act:
        if action.text:
            print(action.text)
    await ask_question(copilot_client, action.conversation.id)


async def main():
    """メイン関数 - コマンドライン引数でCSVファイルが指定された場合は自動処理"""
    if len(sys.argv) > 1:
        csv_file_path = sys.argv[1]
        if Path(csv_file_path).exists():
            await process_csv_questions(csv_file_path)
        else:
            print(f"CSVファイルが見つかりません: {csv_file_path}")
            sys.exit(1)
    else:
        print("使用方法: python main.py <csvファイルパス>")
        print("例: python main.py questions.csv")
        print("\nまたは、インタラクティブモードで実行する場合は引数なしで実行してください")
        
        # CSVファイルが指定されていない場合、インタラクティブモードで実行
        await main_interactive()


if __name__ == "__main__":
    asyncio.run(main())

6. テスト用CSVファイルの作成

一括クエリ実行用のCSVファイルを作成します。以下のような形式でquestions.csvという名前で保存します。内容はお好みで変更してください。

question
こんにちは
あなたは誰ですか?
日本の首都はどこですか?

7. 一括クエリ実行

以下のコマンドで、CSVファイルに記載された質問を一括でエージェントに送信し、回答を取得します。

python -m src.main questions.csv

実行すると、各回答結果がanswer列に追加された形でanswered_questions.csvという名前で保存されます!

まとめ

本記事では、Microsoft 365 Agents SDKを利用してCopilot Studioのエージェントに対してコマンドラインから一括でクエリを実行し、回答を取得する方法を紹介しました。この記事では単純に質問と回答のやり取りまでを行いましたが、得られた回答を用いて例えば以下のようなことが可能になります。

  • Copilot Studioを用いたRAGチャットボットに対し、RAGASなどの評価基準を用いた定量評価を実施する
  • CI/CDパイプラインに組み込み、モデルやプロンプト、データソースの変更が回答精度に与える影響を継続的に監視する
  • Langfuseなどの外部ツールと連携し、より高度なLLMOpsを実現する

上記のような本格的なLLMOpsの実現に向けた第一歩として、本記事が参考になれば幸いです。さらなる運用本格化に向けたご相談も弊社にて承っておりますので、お気軽にお問い合わせください。