1. はじめに
株式会社 neoAI 26新卒入社の上原将磨です。AI ソリューション事業部でAIプロダクトの開発に携わっています。
突然ですが、Slackでこんなふうにタスクを依頼されること、ありませんか？
こういうときに「このままNotionに自動で入ってくれたらいいのに」と思っていました。
そこで開発したのがこのTask Botです。
スレッドでメンションするだけでNotionにタスクの内容/期限/責任者を入力してくれます。
こうした「AIでできそう」「アプリにできそう」という構想はよく浮かびますが、実際に形にするのは意外と難しいものです。
実際に作ってみると、精度の問題、実際の利用シーンとのギャップ、運用コスト、リソース構成の選択や構築など──さまざまな壁にぶつかります。
私は、AIの最新動向を追うことはもちろん、「現場で実際に使われる仕組み」へと落とし込むことまでがAIエンジニアの役割だと考えています。
そこで今回は、開発したアプリケーションの仕組みと、そのの過程で直面した課題、そして取ったアプローチについてご紹介します。
2. 技術構成と実装概要
ここでは、SlackとNotionを連携させるAI Botの技術構成および実装概要について説明します。
2-1. infra構成
Slackでのメンションをトリガーに Azure Functions が動作し、LLM による解析を経て Notion にタスクを登録します。
開発後もニーズに応じて細かくアップデートを重ねたり、将来的にはアプリケーションの拡張や運用を他のメンバーに引き継ぐことも想定して、
リソースは IaC で管理し、GitHub Actions による CI/CD パイプラインを構築しました。
以下実装の際に参考にしたものです。
2-2. アプリケーション構造
Slackからのメンションイベントをトリガーに、簡単なHTTPヘッダーなどの検証を経て、アプリケーションが起動します。
@bp.function_name("SlackWebhook")
@bp.route(route="slack_webhook", auth_level=func.AuthLevel.ANONYMOUS)
def slack_webhook(req: func.HttpRequest) -> func.HttpResponse:
req_body: dict[str, Any] = req.get_json()
header: dict[str, str] = req.headers
raw_request_body: bytes = req.get_body()
manage_task_controller = injector.get(IManageTaskController) # type: ignore[type-abstract]
response = manage_task_controller.handle_event(req_body, header, raw_request_body)
return func.HttpResponse("Success", status_code=200)
アプリケーション内では、slackのスレッド履歴を取得し、それをLLMサービスにを渡し、会話文脈を解析してタスク候補を生成します。
生成されたタスクは Notion の TaskDB に assignee, 期限, Project名付きで保存され、最後にSlack上で結果を返信するまでが1サイクルの処理フローです。
下図はそのシーケンスを示したものです。
Slack のユーザーと Notion のユーザー（Task の assignee）を照合したり、
LLM が予測した Project 名が TaskDB のプロパティ内に存在するかを確認したり、
現在の処理状況をスレッド内でユーザーに通知したりと、
Slack と Notion の API を相互に呼び出しながら処理を行っています。
2-3. LLMへの入力データ
Slackスレッドの内容は、そのままLLMに渡しても「誰が・どの文脈で発言したのか」が不明瞭になりやすいため、各発話を明示的にタグ付けしてテキスト化しています。
具体的には、スレッド履歴を history_turn（過去の会話）と current_turn（最新の発言）に分け、
それぞれを [会話履歴 i - UserID:x] のような形式で整形してLLMに入力しています。
[会話履歴 1 - UserID:U01XXXX]
このタスク、いつまでに対応すればいい？
[会話履歴 2 - UserID:U02XXXX]
来週のミーティングまでにお願い！
[最新の会話 - UserID:U01XXXX]
@bot ↑のタスクを追加して！
3. 発生した問題とその対策
問題1. Slackイベントの重複リクエスト
重複リクエストとは
Slackの Events API では、アプリがイベントを受け取った際に 3秒以内にHTTP 2xxレスポンスを返す ことが求められています。
もしこの応答が間に合わなかった場合、Slackは「イベント配信が失敗した」と判断し、
最大3回まで再送 を行う仕様になっています。
Your app should respond to the event request with an HTTP 2xx within >three seconds.
If it does not, we’ll consider the event delivery attempt failed.
After a failure, we’ll retry three times, backing off exponentially.
— Slack Events API: Responding to events
この仕組みにより、通信エラーや一時的な処理遅延が発生してもイベントが失われないよう保証されています。
一方で、Azure FunctionsのCold Start や一時的な負荷によって3秒以内の応答が難しい場合、
同じイベントが複数回送信されることになります。
その結果、アプリ側で冪等性（同一イベントを一度だけ処理する仕組み）を担保していないと、
同じメンションが複数回処理されてしまう — つまり、Notion上で同じタスクが何度も登録される問題が起こり得ます。
【対策1：】
SlackのEvents APIでは、再送リクエストの際に専用のHTTPヘッダーが付与されます。
そのため、X-Slack-Retry-Num や X-Slack-Retry-Reason を確認することで、重複リクエストを判定できます。
“With each retry attempt, you’ll also be given a x-slack-retry-num HTTP >header indicating the attempt number.”
— Slack Events API: Retries
※しかし、使用している Function as a Service が Cold Startを伴う場合、最初のリクエストを正しく処理できず、結果的に何も動作しなくなってしまうことがあります。
そのため、対策２↓の方法を行なっています。
【対策2：】
リクエストごとに一意のIDを生成し、DBでそのIDを確認することで、重複したリクエストかどうかを判定する方法もあります。
今回は Azure Table Storage を使って重複チェックを行っています。
Azure Table Storage は、NoSQL キー・バリュー型データベース で、スキーマレスで柔軟にデータを格納でき、テーブルは「PartitionKey」と「RowKey」の組み合わせで一意に識別されます。
大量データを低コスト・高スケーラビリティで保存できるのが特徴です。
TaskBotでは、イベントごとにSlack の channel と ts（タイムスタンプ）を組み合わせて一意なIDを生成し、すでに処理済みかどうかをテーブルで管理するシンプルな仕組みにしました。
event_id = f"{app_mention_event.channel}_{app_mention_event.ts}"
Azure Table Storage は超低コストで使えるのも魅力です👇
| 月間イベント数 | 概算コスト（JPY） |
|---|---|
| 10万件 | 約 ¥1 |
| 100万件 | 約 ¥10 |
| 1,000万件 | 約 ¥100 |
保存とトランザクションを合わせてもほぼ無料レベルで、冪等処理を安全かつ安価に実現できます。
問題2. Azure Function の Cold Start
Cold Startとは
サーバーレス環境で、しばらく呼び出されていなかった関数が再びリクエストを受けた時、
インスタンスを新たに立ち上げる／ランタイムが初期化されるなどの処理が必要になり、
起動遅延（レイテンシ）が増える現象を「Cold Start」と言います。 
今回使用したAzureの Consumption プランでは、使用が無いと自動でスケールダウン（0インスタンス）となるため、次のリクエスト時に Cold Start が発生しやすくなります。 
Cold Startの対策
今回は、Azure Functions の Timer Trigger を利用したウォームアップ処理によって、コールドスタート対策を行いました。
5分に1回起動して、loggerを使うことで、自動でスケールダウンを防いでいます。
@bp.function_name(name="TimerWarmup")
@bp.schedule(
schedule="0 */5 * * * *", # 毎時0分、5分、10分…（5分間隔）
arg_name="my_timer",
run_on_startup=True, # 起動時にも一度実行
use_monitor=False, # Durable モニタリング不要
)
def warmup(my_timer: func.TimerRequest) -> None: # noqa: ARG001
logger.info("🌡️ Warmup timer triggered")
値段感（今回＝一番安い Consumption プラン）
- 実行回数：5分おき → 12回/時 × 24時 × 30日 ≒ 8,640回/月
- Azure Functions の無料枠：月100万実行＋400,000 GB-s（Consumption）
→ ウォームアップ 8,640回は 実行回数の無料枠内。処理内容が軽量なら GB-s もまず無料枠内。
重複リクエスト対策とコールドスタート対策を上記の方法で組み合わせることで、堅牢かつ低コストな運用を実現できます。
問題3. Notion APIの型がない問題
Notion APIについて
Pythonでは、Notionの公式SDK（notion-clientなど）は提供されていますが、返ってくるレスポンスに型定義がないため、実質的に HTTP リクエストを自分で投げて JSON を扱うしかありません。
当然ながら、返ってくるレスポンスにも型定義はなく、
構造はワークスペースの設定やデータベース構成に強く依存します。
そのため、型安全な開発が難しいという課題があります。
notionの型安全のための対策
Notion では、弊社の AI エンジニアである Masafumi Higashi が開発した notion-py-client を利用しています。
これは Pydantic v2 を用いた型安全な Notion API クライアントライブラリで、
非同期処理（httpx）に対応し、最新の Notion API（2025-09-03版）をサポートしています。 Notion のレスポンスをドメインモデルに直接マッピングできます。
4. 導入後の変化
現在、この仕組みは実際に社内で利用されており、多くのメンバーに日常的に使ってもらえるようになっています。
また、どんなサービスでも標準的な使い方のままではなく、自社の業務に合わせて自由にカスタマイズできる点が大きな価値だと感じています。
通常であれば「良いDXツールがないか」を外部から探すものですが、自社でAPIを活用して自分たちのワークフローに合わせて作り込んでいく方が、低コスト高インパクトなツールになると実感しています。
さらに、この取り組みをきっかけに、1on1リマインドBotなどの新しい社内ツール開発も始まり、現在では、「社内DXプロジェクト」として継続的に展開しています。
単発のBot開発で終わるのではなく、部署横断的に課題を拾い上げ、それぞれの業務フローやナレッジを合わせてツールを内製する動きが広がっています。
まず自分たち自身の業務環境をスマートにし、“身の回りがすでにDXされている状態” を実現できているからこそ、クライアントに対しても、実体験に裏付けられた高品質なデリバリーを行うことができています。
今回のBot開発は、“あったらいいよね”という構想を実際に使われる仕組みへと落とし込む文化が浸透し始めたことを示す、ひとつの象徴だと考えています。
Discussion