LoginSignup
1
1

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@BNR-Longin株式会社BinaryTech | 日本品質 × ベトナム開発、最高のバランスを。

SlackでAIを強化する | 第3章：会話分析：AIによるインサイト生成

Posted at

はじめに

第2章では、Slackにメッセージを送信し、チャンネルを作成するMCPサーバーを構築し、コミュニケーション自動化エージェントを実現しました。これにより、AIがメッセージ操作やチャンネル管理を処理できるようになり、チームの効率が向上しました。今回は、この基盤を活用して、Slackの会話データを解析する会話分析エージェントを構築します。

この第3章では、Slackチャンネルのメッセージを分析し、コミュニケーション頻度、主要トピック、ユーザーエンゲージメントなどのインサイトを生成します。たとえば、AIが「チームがプロジェクト計画に集中している」ことを検出したり、「特定のユーザーが会話に積極的でない」ことを指摘したりできます。コード例とステップごとのガイドで、会話分析AIの構築を体験しましょう。さあ、始めましょう！

会話分析エージェントとは？

会話分析エージェントは、Slackのメッセージデータを解析し、チームのコミュニケーションに関するインサイトを提供するAIです。MCPサーバーを介して、以下のような機能を実現できます：

  • 頻度分析：チャンネルごとのメッセージ投稿頻度を計算。
  • トピック抽出：メッセージから主要なキーワードやテーマを特定。
  • エンゲージメント分析：ユーザーごとの投稿数や反応（リアクション、返信）を評価。

ユースケース

  • チームパフォーマンス：AIがコミュニケーションの活発さや参加度を分析。
  • プロジェクト管理：AIが会話からプロジェクトの優先事項や課題を抽出。
  • プロセス改善：AIがコミュニケーションのボトルネック（例：特定のトピックが無視される）を特定。

開発環境の準備

第2章の環境を基に、以下の準備を行います：

  • Python 3.8以降、mcpライブラリ、requestsライブラリ、Claude Desktop：これまでと同じ。
  • python-dotenv：環境変数の管理（既にインストール済み）。
  • Slackチャンネル：分析用のメッセージデータを含むチャンネル。

Slackのセットアップ

  1. チャンネル準備：
    • 第2章のチャンネル（例：#mcp-project）を使用。
    • メッセージを追加（例：プロジェクト計画、コードレビュー、タスク割り当てに関する会話）。
    • ユーザーエンゲージメントを高めるため、リアクション（絵文字）やスレッド返信を追加。
    • チャンネルIDを記録（第1章で取得済み）。
  2. Slackアプリの確認：
    • 第2章のアプリを再利用（スコープ：channels:history、channels:read）。
    • 必要に応じてreactions:read（リアクション取得）を追加。
    • アプリがチャンネルに招待済みであることを確認。
  3. 環境変数：
    第2章の.envファイルに以下を確認： 
    SLACK_TOKEN=your_bot_token
SLACK_CHANNEL_ID=your_channel_id

コード例：会話分析用MCPサーバー

以下のMCPサーバーは、Slackチャンネルのメッセージを取得し、頻度やエンゲージメントを分析します。

from mcp import MCPServer
import os
from dotenv import load_dotenv
import requests
from datetime import datetime, timedelta
import statistics
from collections import Counter

class SlackAnalysisServer(MCPServer):
    def __init__(self, host, port, token, channel_id):
        super().__init__(host, port)
        self.token = token
        self.channel_id = channel_id
        self.base_url = "https://slack.com/api"
        self.headers = {
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/json"
        }
        self.register_resource("analyze_conversation", self.analyze_conversation)

    def get_messages(self, limit=100):
        try:
            url = f"{self.base_url}/conversations.history"
            query = {"channel": self.channel_id, "limit": limit}
            response = requests.get(url, headers=self.headers, params=query)
            response.raise_for_status()
            return response.json()["messages"]
        except Exception as e:
            return {"status": "error", "message": str(e)}

    def get_reactions(self, message_ts):
        try:
            url = f"{self.base_url}/reactions.get"
            query = {"channel": self.channel_id, "timestamp": message_ts}
            response = requests.get(url, headers=self.headers, params=query)
            response.raise_for_status()
            return response.json().get("message", {}).get("reactions", [])
        except Exception as e:
            return []

    def analyze_conversation(self, params):
        try:
            messages = self.get_messages(limit=params.get("limit", 100))
            if isinstance(messages, dict) and "status" in messages:
                return messages

            user_counts = Counter()
            reaction_counts = Counter()
            timestamps = []
            keywords = Counter()
            stop_words = {"を", "は", "に", "の", "が", "です", "ます"}

            now = datetime.now()
            for msg in messages:
                user = msg.get("user", "unknown")
                text = msg.get("text", "")
                ts = float(msg["ts"])
                timestamp = datetime.fromtimestamp(ts)

                # ユーザーエンゲージメント
                user_counts[user] += 1

                # リアクション分析
                reactions = self.get_reactions(msg["ts"])
                for reaction in reactions:
                    reaction_counts[reaction["name"]] += reaction["count"]

                # メッセージ頻度
                timestamps.append(timestamp)

                # キーワード抽出（簡易）
                words = text.split()
                for word in words:
                    if word not in stop_words and len(word) > 1:
                        keywords[word] += 1

            # 頻度分析
            if timestamps:
                time_diffs = [(timestamps[i] - timestamps[i+1]).total_seconds() / 3600 
                             for i in range(len(timestamps)-1)]
                avg_interval = statistics.mean(time_diffs) if time_diffs else 0
                messages_per_day = len(timestamps) / ((now - min(timestamps)).total_seconds() / 86400)
            else:
                avg_interval, messages_per_day = 0, 0

            # トップキーワード
            top_keywords = [{"word": w, "count": c} for w, c in keywords.most_common(5)]

            return {
                "status": "success",
                "analysis": {
                    "message_count": len(messages),
                    "avg_message_interval_hours": f"{avg_interval:.2f}",
                    "messages_per_day": f"{messages_per_day:.2f}",
                    "user_engagement": [{"user": u, "count": c} for u, c in user_counts.most_common(3)],
                    "top_reactions": [{"reaction": r, "count": c} for r, c in reaction_counts.most_common(3)],
                    "top_keywords": top_keywords
                }
            }
        except Exception as e:
            return {"status": "error", "message": str(e)}

if __name__ == "__main__":
    load_dotenv()
    server = SlackAnalysisServer(
        host="localhost",
        port=8124,
        token=os.getenv("SLACK_TOKEN"),
        channel_id=os.getenv("SLACK_CHANNEL_ID")
    )
    print("Slack分析MCPサーバーを起動中: http://localhost:8124")
    server.start()

コードの説明

  • get_messages：チャンネルのメッセージを取得（最大100件）。
  • get_reactions：メッセージごとのリアクションを取得。
  • analyze_conversation：メッセージを解析し、以下のインサイトを生成：
    • メッセージ総数：取得したメッセージの数。
    • 平均投稿間隔：メッセージ間の平均時間（時間単位）。
    • 1日あたりのメッセージ数：会話の活発さを評価。
    • ユーザーエンゲージメント：投稿数が多い上位3ユーザー。
    • トップリアクション：使用頻度の高いリアクション（例：👍）。
    • トップキーワード：メッセージから頻出単語を抽出（簡易トピック分析）。
  • register_resource：会話分析をリソースとして登録。

前提条件

  • Slackチャンネルにメッセージとリアクションが存在。
  • Slackアプリにchannels:history、channels:read、reactions:readスコープが付与されている。
  • .envファイルに正しいSLACK_TOKENとSLACK_CHANNEL_IDが設定済み。

サーバーのテスト

サーバーが正しく動作するか確認します：

  1. サーバー起動：

    python slack_analysis_server.py

    コンソールに「Slack分析MCPサーバーを起動中: http://localhost:8124」と表示。

  2. 会話分析のテスト：
    Pythonでリクエストを送信：

    import requests
import json

url = "http://localhost:8124"
payload = {
    "jsonrpc": "2.0",
    "method": "analyze_conversation",
    "params": {"limit": 50},
    "id": 1
}
response = requests.post(url, json=payload)
print(json.dumps(response.json(), indent=2, ensure_ascii=False))

    期待されるレスポンス：

    {
  "jsonrpc": "2.0",
  "result": {
    "status": "success",
    "analysis": {
      "message_count": 50,
      "avg_message_interval_hours": "2.50",
      "messages_per_day": "10.00",
      "user_engagement": [
        {"user": "U123456", "count": 20},
        {"user": "U789012", "count": 15},
        {"user": "U345678", "count": 10}
      ],
      "top_reactions": [
        {"reaction": "thumbsup", "count": 25},
        {"reaction": "smile", "count": 10}
      ],
      "top_keywords": [
        {"word": "プロジェクト", "count": 8},
        {"word": "タスク", "count": 6},
        {"word": "コード", "count": 5}
      ]
    }
  },
  "id": 1
}

Claude Desktopとの接続

サーバーをClaude Desktopに接続します：

  1. 設定ファイルの編集：
    Claude Desktopの設定ファイル（例：claude_desktop_config.json）に以下を追加：

    {
  "mcp_servers": [
    {
      "name": "SlackAnalysisServer",
      "url": "http://localhost:8124",
      "auth": "none"
    }
  ]
}

  2. Claudeでテスト：
    Claude Desktopを起動し、プロンプトを入力：

    チャンネル #mcp-project の会話データを分析してください。

    レスポンス例：

    チャンネル #mcp-project の会話分析：
- メッセージ数：50
- 平均投稿間隔：2.50時間
- 1日あたりのメッセージ数：10.00
- アクティブユーザー：U123456（20投稿）、U789012（15投稿）、U345678（10投稿）
- 人気のリアクション：👍（25）、😊（10）
- 主要トピック：プロジェクト（8）、タスク（6）、コード（5）

実装のコツと注意点

  • データ品質：メッセージが少ない場合、分析結果が制限される。十分なデータ（例：50件以上）を用意。
  • レートリミティング：Slack APIの制限（例：conversations.historyは1秒に1リクエスト）に注意。
  • セキュリティ：本番環境では、auth: noneを避け、トークン認証を導入。
  • テスト：テスト用チャンネルを作成し、本番データに影響を与えない。
  • 拡張性：大量のメッセージを処理する場合、キャッシュ（例：Redis）やNLPライブラリ（例：spaCy）でキーワード抽出を強化。

試してみよう：挑戦課題

以下の機能を追加して、エージェントを強化してみてください：

  • 特定ユーザー（例：U123456）のメッセージだけを分析するフィルター。
  • メッセージの感情分析（ポジティブ/ネガティブ）を行う機能。
  • 分析結果をチャンネルに投稿するツール。

まとめと次のステップ

この第3章では、Slackの会話データを活用して会話分析エージェントを構築しました。メッセージ頻度やエンゲージメントを分析することで、AIがチームのコミュニケーションを評価し、プロセス改善を支援できるようになりました。

次の第4章では、SlackのWebhookを活用してリアルタイム管理AIを構築します。たとえば、AIが新しいメッセージやリアクションをリアルタイムで検知し、通知や応答を生成します。リアルタイム管理AIに興味がある方は、ぜひお楽しみに！

役に立ったと思ったら、「いいね」や「ストック」をしていただけると嬉しいです！次の章でまたお会いしましょう！

1
1
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
1
1

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?