Chatwork API × AIエージェントでチームメンバーの状況を分析する
作成日:
更新日:
以前の記事で、ローカルLLMを使った情報収集・要約・RAGシステムの設計について書きました。
今回はその実践編として、Chatworkに絞って、チーム内のやり取りを定期的に収集し、AIエージェントにチームメンバーの状況を分析させる実験を行いました。
やりたいこと
チャットツールには日々膨大なやり取りが蓄積されています。しかし、その内容を振り返って「誰が何に困っているか」「どのプロジェクトが滞っているか」を把握するのは大変です。
AIに聞けば一発でわかるのでは？
というわけで、以下を試みました：
- Chatwork APIでメッセージを定期収集
- ローカルDBに蓄積
- AIエージェントに「このメンバーの困りごとを分析して」と依頼
- レポートを自動生成
⚠️ セキュリティに関する重要な注意
クラウドAIサービス利用時のリスク
業務チャットには、クライアント情報や機密情報が含まれていることがほとんどです。
これらのデータをChatGPT、Claude、Geminiなどのメジャーな生成AIサービスに送信する場合、以下のリスクがあります：
- データがクラウドに送信される: 機密情報が外部サーバーに渡る
- 学習データに利用される可能性: サービスによっては入力データが学習に使われる
- 情報漏洩のリスク: セキュリティインシデント時に流出する可能性
- コンプライアンス違反: 顧客データの取り扱い規約に抵触する恐れ
本プロジェクトではローカルLLMを使用
上記のリスクを回避するため、本プロジェクトでは**ローカルLLM（Ollama）**を使用しています。
# config/llm.yaml
llm:
provider: ollama
model: mistral # ローカルで動作
base_url: http://localhost:11434
ローカルLLMのメリット:
| 項目 | クラウドAI | ローカルLLM |
|---|---|---|
| データ送信先 | 外部サーバー | 自分のPC内 |
| 学習への利用 | サービスによる | なし |
| オフライン動作 | 不可 | 可能 |
| コスト | 従量課金 | 無料 |
| 機密情報の扱い | 要注意 | 安全 |
業務データをAIで分析する際は、必ずデータの取り扱いに注意してください。
Chatwork API の基本
API の特徴と制限
Chatwork API には以下の特徴があります：
| 項目 | 内容 |
|---|---|
| 取得可能件数 | 1ルームあたり最新100件 |
| 過去データ | 全件取得は不可 |
| レート制限 | 5分間に300リクエスト |
| 認証 | APIトークン（ヘッダー認証） |
重要: Chatwork APIでは過去のデータを全件取得することはできません。日々定期的にAPIを叩いて、差分を蓄積していく必要があります。
全データをダウンロードしたい場合は、エンタープライズプランで過去データのエクスポート機能が利用可能です。
基本的な使い方
import requests
CHATWORK_API_BASE = "https://api.chatwork.com/v2"
API_TOKEN = "your_api_token"
headers = {"X-ChatWorkToken": API_TOKEN}
# ルーム一覧を取得
rooms = requests.get(f"{CHATWORK_API_BASE}/rooms", headers=headers).json()
# 特定ルームのメッセージを取得（最新100件）
room_id = 123456789
messages = requests.get(
f"{CHATWORK_API_BASE}/rooms/{room_id}/messages",
headers=headers,
params={"force": 1} # 未読関係なく取得
).json()
APIの注意点
- force=1 が必要: デフォルトでは未読メッセージのみ取得。
force=1で既読含め最新100件を取得 - 204 No Content: 新着メッセージがない場合、204が返る（JSONではない）
- レート制限: 429エラー時は
Retry-Afterヘッダーを確認して待機
作成したプログラム
プロジェクト構成
punk_records/
├── config/
│ ├── collectors.yaml # 収集設定
│ ├── filters.yaml # フィルタリング設定
│ └── llm.yaml # LLM設定
├── data/
│ ├── vectors/ # ChromaDB（ベクトル検索）
│ ├── index/ # Whoosh（全文検索）
│ └── metadata.db # SQLite（メタデータ）
├── src/punk_records/
│ ├── collectors/
│ │ ├── chatwork.py # Chatwork収集
│ │ ├── slack.py # Slack収集
│ │ └── ...
│ ├── storage/
│ │ ├── vector_store.py # ベクトルDB
│ │ └── metadata_store.py # メタデータ管理
│ ├── search/
│ │ └── hybrid.py # ハイブリッド検索
│ └── qa/
│ └── rag_engine.py # RAG質問応答
└── README.md
ChatworkCollector の実装
差分取得に対応したChatworkコレクターを実装しました：
class ChatworkCollector(BaseCollector):
"""Chatwork API を使ってメッセージを収集するコレクター"""
def __init__(self) -> None:
super().__init__(CollectorContext(SourceType.CHATWORK))
self.api_token = os.getenv("CHATWORK_API_TOKEN", "")
config = self.context.config
# 期間制限（日数）: 0 または未指定で無制限
self.days_back: int = config.get("days_back", 0)
# 差分取得モード: true で前回取得位置以降のみ取得
self.incremental: bool = config.get("incremental", True)
def collect(self) -> Iterable[MessageRecord]:
"""全ルームからメッセージを収集"""
rooms = self.get_rooms()
for room in rooms:
room_id = room["room_id"]
room_name = room.get("name", str(room_id))
# 差分取得: 前回の最終位置を取得
if self.incremental:
last_message_id, last_timestamp = self._get_last_state(room_id)
messages = self.get_messages(room_id)
for msg in messages:
# 差分取得: 前回取得済みのメッセージはスキップ
if self.incremental and last_timestamp:
if msg["send_time"] <= last_timestamp:
continue
yield self._build_record(msg, room_name)
# 最終位置を更新
if new_messages:
self._update_state(room_id, last_msg_id, last_timestamp)
差分取得の仕組み
metadata.db（SQLite）に各ルームの最終取得位置を保存し、次回実行時はそれ以降のメッセージのみを取得します：
def _get_last_state(self, room_id: int) -> Tuple[Optional[str], Optional[int]]:
"""指定ルームの最終取得位置を取得"""
state = self.metadata_store.get_collector_state("chatwork", str(room_id))
if state:
return state.get("last_message_id"), state.get("last_timestamp")
return None, None
def _update_state(self, room_id: int, last_message_id: str, last_timestamp: int) -> None:
"""最終取得位置を更新"""
self.metadata_store.update_collector_state(
"chatwork", str(room_id), last_message_id, last_timestamp
)
設定ファイル（collectors.yaml）
collectors:
chatwork:
enabled: true
mode: api
days_back: 30 # 直近30日分のみ取得（0で無制限）
incremental: true # 差分取得（前回以降の新規メッセージのみ）
room_ids:
- 123456789 # プロジェクトA
- 987654321 # プロジェクトB
# exclude_rooms: [111111111] # 除外するルームID
CLIでの実行
# Chatworkからデータ収集
uv run python -m punk_records.cli collect --source chatwork
# 検索
uv run python -m punk_records.cli search "進捗確認" --mode hybrid
# 質問応答（RAG）
uv run python -m punk_records.cli ask "今週の決定事項は？"
AIエージェントによる分析
分析の依頼方法
蓄積したデータを元に、Cursor（AIエージェント）に以下のように依頼しました：
Chatworkのメッセージから、Aさんが困っていることや
難しいと感じている点を分析してください。
分析対象: 2025年11月〜12月のメッセージ
出力形式: Markdown
実際に生成されたレポート（サンプル）
AIエージェントが生成したレポートの一部です（個人名・案件名は変更しています）：
# Aさんの困りごと分析
## 概要
Chatworkのメッセージから、Aさんが困っていることや
難しいと感じている点を分析したドキュメントです。
**分析期間**: 2025年11月〜12月のメッセージ
**分析対象**: Aさんの発言と、彼への依頼・質問
---
## 1. 情報待ちによる作業停滞
### 具体的な事例
#### プロジェクトX
- **状況**: サーバー情報、GA情報などの提供待ち
- **発言**: 「全然情報来ないね、、！」
- **影響**: 作業が進められない状態が続いている
### 困りごと
- **先方からの情報提供が遅い**: 作業を進められない状態が続く
- **スケジュール調整の難しさ**: 情報待ちのため、正確な見積もりが困難
---
## 2. 複数案件の並行進行による負担
### 勤怠報告から見える進行中の案件
- プロジェクトA：TOP改修、WP実装
- プロジェクトB：結果発表ページ
- プロジェクトC：11月対応
- プロジェクトD：ブラックフライデー対応
- プロジェクトE：スライド画像のCMS化
### 困りごと
- **案件数の多さ**: 10件以上の案件を同時進行
- **優先順位の判断**: どの案件を優先すべきか判断が難しい
- **未完了案件の蓄積**: 「未完了」と記載されている案件が複数
---
## 3. 工数算出の頻繁な依頼
### 困りごと
- ほぼ毎日のように工数算出を求められている
- 詳細な仕様が不明な状態での工数算出
- 修正依頼が多く、再計算が必要
---
## 4. 緊急対応・タイトなスケジュール
### 具体的な事例
- 「本日早めに先方に回答を行いたく、確認いただくこと可能でしょうか」
- 「いつもタイトなお願いとなって申し訳ございません」
### 困りごと
- 緊急対応が頻繁
- 確認待ちでも時間がかかる
- 他の案件のスケジュールが崩れる
---
## 推奨されるサポート
### 情報管理の改善
- 情報待ち案件の可視化
- 情報提供の催促フローの確立
### 工数算出の効率化
- よくある工数算出のテンプレート化
- 複数の工数算出を一度にまとめて対応
### 案件管理の改善
- 優先順位の明確化
- 未完了案件の定期的な整理
分析の精度
AIは以下の点に着目して情報を抽出していました：
- 感情表現の検出: 「全然情報来ないね」という愚痴から、情報待ちのストレスを検知
- パターンの発見: 毎日の勤怠報告から、同時進行している案件数を集計
- 定型句の認識: 「いつもタイトなお願い」という表現から、緊急対応の多さを推測
- 改善提案: 困りごとに対する具体的なサポート案を提示
技術的なポイント
ハイブリッド検索の活用
メンバー分析では、セマンティック検索とキーワード検索のハイブリッドをAIが効果的に利用しているようでした：
def hybrid_search(query: str, top_k: int = 50):
# セマンティック検索（意味で検索）
semantic_results = vector_store.query(query, n_results=top_k)
# キーワード検索（人名・案件名）
keyword_results = keyword_index.search(query, limit=top_k)
# 結果を統合
return merge_results(semantic_results, keyword_results)
理由:
- セマンティック検索: 「困っている」「難しい」などの感情表現を拾う
- キーワード検索: 人名、案件名を正確にマッチ
今後の展望
定期レポートの自動化
まだまだ実用には足らないレベル感かもしれませんが、メンバーの状況の概要や問題点を見つけるためのとっかかりにはなるかもしれないと感じました。 現在は手動でAIに依頼していますが、将来的に下記のようなレポートの自動化も検討したいです。
- 毎週金曜に自動でチーム状況レポートを生成
- 「困りごとスコア」が高いメンバーをアラート
- 滞留している案件を自動検出
他サービスとの統合
Chatworkだけでなく、以下も統合予定：
- Slack: エンジニアチームのやり取り
- Gmail: クライアントとのメールやり取り
- GitHub: Issue/PRの状況
まとめ
Chatwork API × AIエージェントで、以下を実現しました：
- 日々の差分取得: APIの制限を考慮した定期収集
- ローカルDB蓄積: ChromaDB + SQLiteでデータ管理
- AIによる分析: メンバーの困りごとを自動抽出
- レポート生成: 改善提案を含むMarkdown出力
ポイント:
- Chatwork APIは過去データを全件取得できないので、日々の定期収集が必須
- 差分取得機能を実装して、効率的にデータを蓄積
- AIは感情表現やパターンを検出し、人間では気づきにくい課題を発見
- 業務データはローカルLLMで処理し、クラウドAIへの送信リスクを回避
大規模なチームの状況把握に悩む人、膨大なチャットログを活用したい人には、ぜひ試してほしいアプローチです。
ただし、クライアント情報を含むデータをAIで分析する際は、必ずセキュリティに配慮してください。クラウドAIサービスを利用する場合は、データの取り扱い規約を十分に確認しましょう。