DatabricksでPowerPointを作成するChatAgentを作る

こちらの精神的続編です。

https://qiita.com/isanakamishiro2/items/ec75947ea9bb62f394d2

# 導入

[Model Context Protocol(MCP)](https://github.com/modelcontextprotocol)を使ったエージェントの第３弾です。

タイトルの通り、公開されているMCP Serverを活用してPowerPointを作成します。

日本企業の多くで資料にPowerPointなどのスライドを利用することが多いのではないかと思います。これをちょっとでも楽にできるエージェントがあると便利ですよね。

というわけで、今回は以下のリポジトリで公開されているPowerPoint MCP Serverを使ってPowerPointを作成するエージェントをDatabricks上で作成します。

https://github.com/GongRzhe/Office-PowerPoint-MCP-Server

:::note info
PowerPoint操作のMCPは他にも公開されているものがあります。
お好みで別のMCP Serverも使ってみてください。
:::


進め方は[前回記事](https://qiita.com/isanakamishiro2/items/ec75947ea9bb62f394d2
)以下とほとんど同じですが、今回はModel Servingでのデプロイはしません。（ファイル作成と取得がやりづらいので）
また、前回利用した[Markitdown-MCP](https://github.com/microsoft/markitdown/tree/main/packages/markitdown-mcp)を今回も利用しています。




実行はDatabricks on AWS、ノートブックはサーバレスクラスタを利用しました。

:::note info
コード飛ばして、実行結果だけ確認したい方はStep3から確認ください。
:::


# Step1. MCPを使うエージェントを定義

まずはMCPサーバをツールとして利用するエージェントを定義します。
ノートブックを作成し、必要なパッケージをインストール。

```python
%pip install langchain-mcp-adapters langgraph databricks-langchain databricks-agents rich nest-asyncio mcp-simple-timeserver markitdown-mcp ffmpeg-python python-pptx
%pip install mlflow

%restart_python
```

前回同様、MCPサーバと連携するためのモジュールとして[LangChain MCP Adapters](https://github.com/langchain-ai/langchain-mcp-adapters)を利用しています。


次に[こちら](https://github.com/GongRzhe/Office-PowerPoint-MCP-Server
)のリポジトリを```git clone```で取得。```/tmp```配下にクローンします。

```python
!cd /tmp && git clone https://github.com/GongRzhe/Office-PowerPoint-MCP-Server.git powerpoint-mcp
!chmod +x /tmp/powerpoint-mcp/ppt_mcp_server.py
```

では、MCPを使うエージェントを定義します。
MLflowのChatAgentインターフェースを備えたカスタムクラスとして定義しています。
[前回記事](https://qiita.com/isanakamishiro2/items/ec75947ea9bb62f394d2
)との違いは、Markitdown-MCPなど、複数のMCP Serverと連携するように処理を組んでいることです。

```python
%%writefile powerpoint_mcp_agent.py

from typing import Literal, Generator, List, Optional, Any, Dict, Mapping, Union
import uuid
import asyncio
import sys
import os

import mlflow
from databricks_langchain import ChatDatabricks
from langgraph.prebuilt import create_react_agent

from langchain_core.messages import BaseMessage, convert_to_openai_messages
from mlflow.pyfunc import ChatAgent
from mlflow.types.agent import (
    ChatAgentChunk,
    ChatAgentMessage,
    ChatAgentResponse,
    ChatContext,
)
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_mcp_adapters.tools import load_mcp_tools
from functools import reduce

# 環境変数
PPT_OUTPUT_DIR = os.environ.get("PPT_OUTPUT_DIR", os.getcwd())

# Tracingの有効化
mlflow.langchain.autolog()


class PowerPointMCPChatAgent(ChatAgent):
    def __init__(self, model):
        """LangGraphのグラフを指定して初期化"""
        self.model = model
        self.mcp_connections = {}

    def load_context(self, context):
        # artifactからPowerPoint-MCP Serverのリポジトリパスを取得
        powerpoint_mcp_path = context.artifacts["powerpoint_mcp"]
        self.mcp_connections = {
            "markitdown": {
                "command": "markitdown-mcp",
                "args": [],
                "transport": "stdio",
            },
            "powerpoint": {
                "command": "python",
                "args": [
                    os.path.join(powerpoint_mcp_path, "ppt_mcp_server.py")
                ],
                "transport": "stdio",
            },
            "simple-timeserver": {
                "command": "python",
                "args": ["-m", "mcp_simple_timeserver"],
                "transport": "stdio",
            },
        }
        print("loaded contexts.")
        print(self.mcp_connections)

    def predict(
        self,
        messages: list[ChatAgentMessage],
        context: Optional[ChatContext] = None,
        custom_inputs: Optional[dict[str, Any]] = None,
    ) -> ChatAgentResponse:
        """
        指定されたチャットメッセージリストを使用して回答を生成する

        Args:
            messages (list[ChatAgentMessage]): チャットエージェントメッセージのリスト。
            context (Optional[ChatContext]): オプションのチャットコンテキスト。
            custom_inputs (Optional[dict[str, Any]]): カスタム入力のオプション辞書。

        Returns:
            ChatAgentResponse: 予測結果を含むChatAgentResponseオブジェクト。
        """

        gen_messages = []
        usages = []
        for c in self.predict_stream(messages, context, custom_inputs):
            gen_messages.append(c.delta)
            usages.append(c.usage.model_dump(exclude_none=True))

        return ChatAgentResponse(messages=gen_messages, usage=self._sum_usages(usages))

    def predict_stream(
        self,
        messages: list[ChatAgentMessage],
        context: Optional[ChatContext] = None,
        custom_inputs: Optional[dict[str, Any]] = None,
    ) -> Generator[ChatAgentChunk, None, None]:
        """
        指定されたチャットメッセージリストを使用して、非同期的にエージェントを呼び出し、結果を取得します。

        Args:
            messages (list[ChatAgentMessage]): チャットエージェントメッセージのリスト。
            context (Optional[ChatContext]): オプションのチャットコンテキスト。
            custom_inputs (Optional[dict[str, Any]]): カスタム入力のオプション辞書。

        Returns:
            ChatAgentResponse: 予測結果を含むChatAgentResponseオブジェクト。
        """

        async def stream_with_mcp_tools(model, connections, messages: list):
            """MCP Serverと通信して回答を取得する"""

            async with MultiServerMCPClient(connections) as client:
                # エージェントを作成して実行
                agent = create_react_agent(model, client.get_tools())
                async for event in agent.astream(
                    {"messages": messages},
                    config={"recursion_limit": 100},
                    stream_mode="updates",
                ):
                    for v in event.values():
                        messages = v.get("messages", [])
                        for msg in messages:
                            yield ChatAgentChunk(
                                delta=self._convert_lc_message_to_chat_message(msg),
                                usage=msg.response_metadata.get("usage", {}),
                            )
                await client.exit_stack.aclose()

        streamer = stream_with_mcp_tools(
            self.model,
            self.mcp_connections,
            self._convert_messages_to_dict(messages),
        )

        loop = asyncio.get_event_loop()
        while True:
            try:
                yield loop.run_until_complete(streamer.__anext__())

            except StopAsyncIteration:
                break
            except Exception as e:
                # for "Attempted to exit cancel scope in a different task than it was entered in" Exception.
                break

    def _convert_lc_message_to_chat_message(
        self, lc_message: BaseMessage
    ) -> ChatAgentMessage:
        """LangChainメッセージをChatAgentMessageに変換する。"""
        msg = convert_to_openai_messages(lc_message)
        if not "id" in msg:
            msg.update({"id": str(uuid.uuid4())})

        return ChatAgentMessage(**msg)

    def _sum_usages(self, usages: list[dict]) -> dict:
        """使用量のリストから使用量を合計する。"""

        def add_usages(a: dict, b: dict) -> dict:
            pt = "prompt_tokens"
            ct = "completion_tokens"
            tt = "total_tokens"
            return {
                pt: a.get(pt, 0) + b.get(pt, 0),
                ct: a.get(ct, 0) + b.get(ct, 0),
                tt: a.get(tt, 0) + b.get(tt, 0),
            }

        return reduce(add_usages, usages)

    def _convert_lc_message_to_chat_message(
        self, lc_message: BaseMessage
    ) -> ChatAgentMessage:
        """LangChainメッセージをChatAgentMessageに変換する。"""
        msg = convert_to_openai_messages(lc_message)
        if not "id" in msg:
            msg.update({"id": str(uuid.uuid4())})

        return ChatAgentMessage(**msg)

    def _sum_usages(self, usages: list[dict]) -> dict:
        """使用量のリストから使用量を合計する。"""

        def add_usages(a: dict, b: dict) -> dict:
            pt = "prompt_tokens"
            ct = "completion_tokens"
            tt = "total_tokens"
            return {
                pt: a.get(pt, 0) + b.get(pt, 0),
                ct: a.get(ct, 0) + b.get(ct, 0),
                tt: a.get(tt, 0) + b.get(tt, 0),
            }

        return reduce(add_usages, usages, {})


# DatabricksネイティブのClaude 3.7 SonnetをLLMとして利用
LLM_ENDPOINT_NAME = "databricks-claude-3-7-sonnet"
llm = ChatDatabricks(model=LLM_ENDPOINT_NAME)

# エージェントを作成
AGENT = PowerPointMCPChatAgent(model=llm)
mlflow.models.set_model(AGENT)
```

これで準備は終わりました。

# Step2. エージェントをロギング

作成したエージェントをMLflow上にロギング（保管）します。

まず```nest_asyncio```パッケージを使って、```async.io```をノートブック内で利用可能にします。

```python
import nest_asyncio

nest_asyncio.apply()
```

次にMLflowの```log_model```を実行して、Step1で定義したカスタムChatAgentクラスを保管します。

```python
import mlflow
from powerpoint_mcp_agent import LLM_ENDPOINT_NAME
from mlflow.models.resources import DatabricksServingEndpoint

resources = [DatabricksServingEndpoint(endpoint_name=LLM_ENDPOINT_NAME)]

input_example = {
    "messages": [
        {
            "role": "user",
            "content": "今何時？",
        }
    ]
}

artifacts = {
    "powerpoint_mcp": "/tmp/powerpoint-mcp/", # git cloneしたPowerPoint-MCPリポジトリのパス
}

with mlflow.start_run():
    logged_agent_info = mlflow.pyfunc.log_model(
        artifact_path ="powerpoint_mcp_agent",
        artifacts=artifacts,
        python_model="powerpoint_mcp_agent.py",
        input_example=input_example,
        pip_requirements=[
            "mlflow",
            "langgraph==0.4.0",
            "langchain-mcp-adapters==0.0.9",
            "databricks-langchain==0.4.2",
            "markitdown-mcp==0.0.1a3",
            "ffmpeg-python==0.2.0",
            "mcp-simple-timeserver==1.0.6",
            "python-pptx==1.0.2",
        ],
        resources=resources,
    )
```

これで作成したエージェントが利用可能になりました。

では、次のStepから実際に使ってみます。

# Step3. エージェントを使ってみる

Step2でロギングしたエージェントをロードします。

```python
import mlflow

agent = mlflow.pyfunc.load_model(logged_agent_info.model_uri)
```

では、このエージェントに対していろいろ指示してみましょう。

```python
agent.predict(
    {
        "messages": [
            {
                "role": "user",
                "content": "LLMの概要解説を文字だけの単純なPowerPoint1枚で作成して",
            }
        ]
    }
)
```

MLflow Tracingの結果は以下のようになります。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/168727/2f6ca2f6-7153-4fd2-93fa-7d3694488218.png)

最終出力内容のように、ノートブックと同じフォルダに「LLM概要解説.pptx」というファイルが作られます。
ダウンロードして開くと、以下のようなPowerPointファイルとなっています。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/168727/7c1089ad-b1ad-4c51-a36b-17ca87a1a44a.png)

文字情報だけですが、コピペでPowerPoint作るよりは楽かな。

さらに、以下のようなクエリも実行してみます。

```python
agent.predict(
    {
        "messages": [
            {
                "role": "user",
                "content": (
                    "次のURLの内容から、Qwen3に関する特徴をPowerPoint3枚でまとめてください。" 
                    "https://qiita.com/isanakamishiro2/items/1104e65c7ebeb5b30c02"
                ),
            }
        ]
    }
)
```

指定したURLは以下の記事になります。

https://qiita.com/isanakamishiro2/items/1104e65c7ebeb5b30c02

実行後のTracing結果は以下の通り。

まず、該当のURLをMarkitdown-MCPでMarkdownに変換して取得する処理が実行されます。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/168727/4a7c190b-8156-41f4-9a56-a29b5d279906.png)


その後、PowerPoint作成のツールが諸所呼び出されたのち、今度は「Qwen3の特徴.pptx」というファイルが作られました。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/168727/a22529f5-8d86-42e4-9f13-0836782d1681.png)

PowerPointファイルを開くと以下のような内容でした。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/168727/821e196c-d687-436f-83df-c6d1c2af4495.png)

このように、他のMCP Serverのツールと連携してPowerPointを作ることが出来ます。
特定のサイトやファイルを使ってレポートを作る際などに重宝しそう。

今回の例では出てきませんでしたが、指示次第で画像の利用や図形・表を使ったPowerPointファイルの作成も可能です。

# おわりに

PowerPointを作成するエージェントを作成してみました。
今回のようにMCPを呼び出すだけだと、実用性は正直まだ低いかなという感想です。
レイアウトも崩れやすいので。

実際にはスライドを前提としたコンテンツの作成や、レイアウト指示の作り込みを行うエージェントシステムを作る必要があり、その最終段階としてPowerPoint形式でファイル出力する、というフローを組む必要があるかなと考えています。
また、入力トークン数がかなり大きくなるので、セルフホストしたLLMを利用するなどコストに対する工夫も必要になりそう。

実用性のあるエージェントシステムを作り込むことができたら、多くの日本企業にとっては救いになる・・・かもしれませんね。