はじめに
Microsoft 365 CopilotのSharePointエージェントは、自然言語でSharePoint内の特定のファイルを検索・要約できる便利なAIアシスタントです。しかし、ライセンスが割り当てられていないユーザーは、SharePointエージェントを作成することができません。詳細は把握できていませんが、ナレッジソースに含めることができるソースの数は20個までなど、色々と制約事項もあるようです。
そこで、似たような機能がDatabricksでも実現できないかと考え、SharePoint Copilot風のエージェント実装を試してみました。
実現したいこと
SharePoint Copilot風の以下の機能を持つエージェントを作成：
- SharePointファイル一覧表示
- キーワードによるファイル検索
- ドキュメント要約
- 自然言語での対話
技術スタック
- プラットフォーム: Databricks
- LLM: databricks managed LLM (databricks-claude-3-7-sonnetなど)
- SharePoint API: Microsoft Graph API
- エージェント構築: MLflow ChatAgent, OpenAI Agents SDK（今回はLLM呼び出しのみ）
- デプロイ: Databricks Mosaic AI Agent Framework
- 認証: Azure AD (MSAL: Microsoft Authentication Library)
アーキテクチャ概要
実装のポイント
※あくまで一例としてご覧ください。実際の環境に合わせて調整いただければと思います。
1. SharePoint Site IDの取得
SharePointにアクセスするためには、まずSite IDを取得する必要があります。ホスト名とサイト名からSite IDを動的に取得する仕組みを実装しました。
import msal
import requests
def get_site_id_by_params(host_name, site_name, tenant_id, client_id, client_secret):
"""ホスト名とサイト名からSite IDを取得
例えばサイトURLが"company_name.sharepoint.com/sites/mysite"の場合,
ホスト名:"company_name", サイト名:"mysite"
"""
app = msal.ConfidentialClientApplication(
client_id=client_id,
client_credential=client_secret,
authority=f"https://login.microsoftonline.com/{tenant_id}"
)
result = app.acquire_token_for_client(scopes=["https://graph.microsoft.com/.default"])
access_token = result["access_token"]
full_hostname = f"{host_name}.sharepoint.com"
graph_url = f"https://graph.microsoft.com/v1.0/sites/{full_hostname}:/sites/{site_name}"
headers = {'Authorization': f'Bearer {access_token}', 'Content-Type': 'application/json'}
response = requests.get(graph_url, headers=headers)
if response.status_code == 200:
site_data = response.json()
return site_data['id'] # これがSite ID
else:
raise Exception(f"Site ID取得エラー: {response.status_code}")
2. SharePointクライアントとREST API接続
Microsoft Graph APIを使用してSharePointにアクセスします。
from pydantic import BaseModel
import requests
class SharePointConfig(BaseModel):
tenant_id: str
client_id: str
client_secret: str
site_id: str
class SharePointClient:
def __init__(self, config: SharePointConfig):
self.config = config
self._access_token = None
async def _get_access_token(self):
"""アクセストークンを取得"""
if self._access_token:
return self._access_token
app = msal.ConfidentialClientApplication(
client_id=self.config.client_id,
client_credential=self.config.client_secret,
authority=f"https://login.microsoftonline.com/{self.config.tenant_id}"
)
result = app.acquire_token_for_client(scopes=["https://graph.microsoft.com/.default"])
self._access_token = result["access_token"]
return self._access_token
async def get_files_list(self, folder_path=None):
"""ファイル一覧を取得"""
try:
token = await self._get_access_token()
if folder_path:
url = f"https://graph.microsoft.com/v1.0/sites/{self.config.site_id}/drive/items/{folder_path}/children"
else:
url = f"https://graph.microsoft.com/v1.0/sites/{self.config.site_id}/drive/root/children"
headers = {'Authorization': f'Bearer {token}', 'Content-Type': 'application/json'}
response = requests.get(url, headers=headers)
if response.status_code == 200:
data = response.json()
files = []
for item in data.get('value', []):
files.append({
'id': item.get('id', ''),
'name': item.get('name', ''),
'size': item.get('size', 0),
'type': 'folder' if item.get('folder') else 'file'
})
return files
else:
return [{'error': f'HTTP {response.status_code}: {response.text}'}]
except Exception as e:
return [{'error': f'ファイル取得エラー: {str(e)}'}]
3. ファイル検索機能
SharePointのフォルダ階層を再帰的に検索する機能を実装しました。
async def search_files_recursive(self, query: str, max_depth: int = 2):
"""再帰的にファイルを検索"""
all_results = []
async def search_folder(folder_id=None, folder_path="root", current_depth=0):
if current_depth >= max_depth:
return
files = await self.get_files_list(folder_id)
if files and 'error' not in files[0]:
# ファイル名にクエリが含まれるものを検索
for file in files:
if query.lower() in file.get('name', '').lower():
file['full_path'] = f"{folder_path}/{file['name']}" if folder_path != "root" else file['name']
all_results.append(file)
# サブフォルダを再帰的に検索
folders = [f for f in files if f.get('type') == 'folder']
for folder in folders:
subfolder_path = f"{folder_path}/{folder['name']}" if folder_path != "root" else folder['name']
await search_folder(folder['id'], subfolder_path, current_depth + 1)
await search_folder()
return all_results
4. Databricks Claude用のクライアント設定
OpenAI互換のAPIを使用してDatabricks Claude (Databricks managed LLM)を呼び出します。
from openai import AsyncOpenAI
# Databricks用のクライアント設定
databricks_client = AsyncOpenAI(
base_url=f'https://{spark.conf.get("spark.databricks.workspaceUrl")}/serving-endpoints',
api_key=dbutils.secrets.get(scope=scope, key="databricks_token"),
)
MODEL = "databricks-claude-3-7-sonnet"
# LLM応答生成での使用例
async def _generate_ai_response(self, user_input: str) -> str:
try:
system_prompt = """あなたはSharePoint Copilotアシスタントです。
利用可能な機能:
🗂️ ファイル一覧の取得 - 「ファイル一覧を表示して」
🔍 ドキュメント検索 - 「[キーワード]を検索して」
📄 ドキュメント要約 - 「[ファイル名]を要約して」
親しみやすく、分かりやすい日本語で回答してください。"""
response = await databricks_client.chat.completions.create(
model=MODEL,
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_input}
],
max_tokens=500,
temperature=0.7
)
return response.choices[0].message.content
except Exception as e:
return f"申し訳ございません。エラーが発生しました: {str(e)}"
5. MLflow ChatAgentによるエージェント統合
各機能を統合してエージェントとして動作させます。
from mlflow.pyfunc import ChatAgent
from mlflow.types.agent import ChatAgentResponse, ChatAgentMessage
class SharePointCopilotAgent(ChatAgent):
def __init__(self):
self.context = SharePointContext()
def _extract_user_input(self, messages):
"""最新のユーザーメッセージを抽出"""
for message in reversed(messages):
if isinstance(message, dict) and message.get("role") == "user":
return message.get("content", "")
elif hasattr(message, 'role') and message.role == "user":
return message.content
return ""
def _create_context(self, context=None):
"""コンテキストを作成"""
sp_context = SharePointContext()
if context:
if isinstance(context, dict):
sp_context.conversation_id = context.get("conversation_id")
sp_context.user_id = context.get("user_id")
else:
if hasattr(context, 'conversation_id'):
sp_context.conversation_id = getattr(context, 'conversation_id', None)
if hasattr(context, 'user_id'):
sp_context.user_id = getattr(context, 'user_id', None)
return sp_context
async def _route_request(self, user_input: str, context) -> str:
"""ユーザーリクエストを適切な機能にルーティング"""
user_input_lower = user_input.lower()
search_keywords = ['検索', 'search', '探す', 'find', '関連']
file_list_keywords = ['ファイル一覧', 'ファイルリスト', 'files', 'list', '一覧を']
if any(keyword in user_input_lower for keyword in search_keywords):
query = self._extract_search_query(user_input)
if query:
return await search_documents(context, query)
else:
return "🔍 検索したいキーワードを教えてください。"
elif any(keyword in user_input_lower for keyword in file_list_keywords):
return await get_sharepoint_files(context)
elif any(keyword in user_input_lower for keyword in ['要約', 'summary', 'summarize']):
file_name = "指定されたファイル"
return await summarize_document(context, file_name)
else:
return await self._generate_ai_response(user_input)
def predict(self, messages, context=None, custom_inputs=None):
user_input = self._extract_user_input(messages)
sp_context = self._create_context(context)
# 非同期処理を実行
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
try:
result = loop.run_until_complete(self._route_request(user_input, sp_context))
finally:
loop.close()
return ChatAgentResponse(
messages=[ChatAgentMessage(role="assistant", content=result, id=str(uuid4()))]
)
実行例
ファイル一覧表示
ユーザー: "SharePointのファイル一覧を表示してください"
エージェント:
📂 SharePointファイル一覧:
📁 **ProjectDocuments** (更新: 2024-12-01 10:30)
📄 **Meeting_Notes.docx** (更新: 2024-11-28 14:15)
📄 **Budget_Report.xlsx** (更新: 2024-11-25 09:45)
📁 **Archive** (更新: 2024-10-15 16:20)
📊 合計: 15 件
ファイル検索
ユーザー: "プロジェクト関連のファイルを検索してください"
エージェント:
🔍 **'プロジェクト'** の検索結果:
📁 **ProjectDocuments**
📍 ProjectDocuments
📄 **Project_Plan_2024.docx**
📍 ProjectDocuments/Project_Plan_2024.docx
📊 検索結果: 12 件
自然言語での対話
ユーザー: "SharePointでどんなことができますか？"
エージェント:
SharePointは、チームでの情報共有やコラボレーションをサポートする強力なプラットフォームです。
## 基本機能
- **ドキュメント管理**: ファイルの保存、共有、バージョン管理
- **チームサイト作成**: プロジェクトやチーム専用の情報共有スペース
- **検索機能**: サイト内のコンテンツを効率的に検索
## 私でサポートできること
- 🗂️ サイト内のファイル一覧表示
- 🔍 キーワードでのドキュメント検索
- 📄 ドキュメントの要約作成
何か具体的にお手伝いできることはありますか？
デプロイとモデル登録
エージェントチャットモデルのファイル保存
MLflow ChatAgentをログに記録し、Databricks Mosaic AI Agent Frameworkでデプロイするためには、エージェントのPythonコード（例: SharePointCopilotAgent クラス、関連するクラスや関数など）を単一のファイルとして保存しておく必要があります。MLflowは、このファイルからエージェントのクラス定義を読み込み、モデルとしてパッケージ化します。
mlflow.pyfunc.log_model で python_model パラメータを使用してエージェントのコードを指定する際は、エージェントクラスの定義を含む全ての関連コードが単一のPythonファイルに記述されている必要があります。
例えば、今回のPythonコードをsharepoint_copilot_agent.pyというファイル名で保存してから、以下のMLflowロギングを実行してください。Databricksノートブックで作業する場合は、%%writefile sharepoint_copilot_agent.py マジックコマンドを使ってコードをファイルに書き出すことができます。
Unity Catalogへの登録
エージェントをUnity Catalogに登録します。
# MLflowでモデルをロギング
with mlflow.start_run():
logged_model_info = mlflow.pyfunc.log_model(
artifact_path="sharepoint_copilot_agent",
python_model="sharepoint_copilot_agent.py",
pip_requirements=[
"mlflow", "msal", "msgraph-sdk",
"pydantic", "openai", "requests", "nest-asyncio"
]
)
# Unity Catalogに登録
UC_MODEL_NAME = "catalog.schema.sharepoint_copilot_agent"
mlflow.register_model(
model_uri=logged_model_info.model_uri,
name=UC_MODEL_NAME
)
Mosaic AI Agent Frameworkでのデプロイ
from databricks import agents
# エンドポイントにデプロイ
agents.deploy(
model_name=UC_MODEL_NAME,
model_version=1,
environment_vars={
# SharePoint認証情報
"SHAREPOINT_TENANT_ID": "{{secrets/scope/tenant_id}}",
"SHAREPOINT_CLIENT_ID": "{{secrets/scope/client_id}}",
"SHAREPOINT_CLIENT_SECRET": "{{secrets/scope/client_secret}}",
# SharePointサイト情報（今回はget_site_id_by_params関数でSite IDを動的に取得）
"SHAREPOINT_HOST_NAME": "your-tenant", # 例: "company_name"
"SHAREPOINT_SITE_NAME": "your-site", # 例: "mysite"
# Databricks LLM呼び出し用
"DATABRICKS_TOKEN": "{{secrets/scope/databricks_token}}",
"DATABRICKS_BASE_URL": "https://your-workspace.databricks.com/serving-endpoints"
}
)
注意事項
Unity Catalog関数と外部接続の推奨
本実装では、SharePoint APIアクセスをPythonクラス内で直接行っています。これは、CREATE CONNECTIONの権限がないユーザーでも柔軟に実装を進めるための一つの方法です。
しかし、Databricksが推奨するベストプラクティスとしては、Unity Catalogの外部接続 (CREATE CONNECTION) を利用してSharePoint APIへの接続情報を一元管理し、それを通じてUnity Catalog関数を実装することです。
これにより、セキュリティ、ガバナンス、再利用性が向上します。
今回の実装は、CREATE CONNECTION権限がない環境 (例えば、企業内のポリシー上、管理ユーザー以外には外部サービスに接続する権限が付与されていない環境）での実装例としてご理解ください。今回のMicrosoft Graph APIでのアクセス方法は十分にセキュアだと考えていますが、Databricksドキュメントのツールを外部サービスに接続するで詳細が紹介されているCREATE CONNECTIONとUnity Catalog関数の利用も必要に応じて検討してください。
セキュリティ考慮事項
以下のセキュリティ事項を必ず確認してください：
- SharePointアクセス権限の適切な設定
- Azure ADアプリケーションの最小権限の原則
- 機密情報の適切な取り扱い
今後の課題と検討事項
今回の実装では基本的な機能を実現できましたが、高度な機能については実装できませんでした。今後の拡張として検討していきたいと思います：
1. ファイル内容の意味的検索
- SharePoint Copilot: ファイル内容を理解して意味的に関連するドキュメントを検索
- 今回の実装: ファイル名のみの検索
- 今後の検討: Databricks Vector Search（Retrieval-Augmented Generation）の実装
2. 高度なドキュメント要約
- SharePoint Copilot: Word、Excel、PowerPointファイルの内容を理解して詳細要約
- 今回の実装: 基本的なメタデータ表示のみ
- 今後の検討: ファイル内容の取得とLLMによる要約機能
3. 複数ファイルにまたがる横断的分析
- SharePoint Copilot: 複数のドキュメントから情報を統合して回答
- 今回の実装: 単一ファイル単位での処理
- 今後の検討: Databricks Vector Search（Retrieval-Augmented Generation）の実装
4. OpenAI Agents SDK / LangGraphを用いたマルチエージェント実装
- SharePoint Copilot: 複数の専門エージェントが連携して複雑なタスクを処理
- 今回の実装: 単一エージェントによる機能ルーティング
-
今後の検討: OpenAI Agents SDKやLangGraphを活用した以下のマルチエージェント構成
- Agent 1: SharePointアクセス専門エージェント（ファイル操作・検索）
- Agent 2: 要約・分析専門エージェント（LLMによる高度な分析）
- Agent 3: ユーザー応答専門エージェント（自然言語生成・対話管理）
- オーケストレーター: エージェント間の連携とワークフロー制御
5. 自然言語によるファイル操作
- SharePoint Copilot: 「新しいSharePointリストを作成して」「このファイルをTeams共有して」「Excelデータを分析して）などの操作
- 今回の実装: 読み取り専用の機能のみ
- 今後の検討: Graph APIの書き込み権限を活用した操作機能
まとめ
今回は試験的な実装でしたが、DatabricksとMicrosoft Graph APIの組み合わせパターンを確立し、今後の高度な機能開発に向けた技術的な知見を得ることができました。Databricks環境を活用したSharePointエージェント風の機能は、有効なアプローチと考えています。
基本機能の範囲ではエージェントの実装が不要な場合もありますが、今後はファイル内容の意味的検索やマルチエージェント構成などの高度な機能に取り組み、より実用的なSharePointエージェントの実現を目指していきます。また、SharePointに限らず、Microsoft Teamsなどの他のサービスとの連携も視野に入れたいところです。
さらに、Lake Flow ConnectやModel Context Protocol (MCP)など、他の有効な手法についても継続的に調査し、最適な実装方法を検討していきたいと考えています。