Telegram Bot + AI 画像認識：フィードバック自動分類システムの構築ガイド

プロダクトのフィードバックが Slack の DM、Line グループ、メールの添付ファイルに散らばっている。最も価値があるのはスクリーンショットだ。エラー画面、UI の崩れ、競合の優れた機能。しかしスマホの画像を Jira に移して説明を書いてラベルを付ける、この手順の摩擦が大きすぎて、フィードバックの半分は「保存済みメッセージ」に埋もれたまま終わる。

自分のサイドプロジェクトでもこの問題に直面していた。毎週 10〜15 件ほどのスクリーンショットがチャット履歴に埋もれ、処理しようと思い出す頃には数日が経過していた。そこで Telegram Bot と LLM Vision API を組み合わせた自動分類システムを構築した。スクリーンショットの受信から構造化された分類結果の出力まで、手動操作は一切不要だ。本記事ではシステム全体をゼロから構築する手順を解説する。

TL;DR

• BotFather で 5 分以内に Bot を作成し、Token と Chat ID を取得
• Python で getUpdates API を使ってメッセージを取得し、テキストと画像を抽出
• LLM Vision API（GPT-4o または Claude）で bug/feature を自動分類
• cron job でスケジューリングし、完全自動化
• 個人利用ならほぼゼロコスト

---

なぜ Telegram Bot なのか？

フィードバック収集のソリューションの中で、Telegram Bot の優位性は明確だ：

ソリューション	設定難易度	費用	モバイル UX	画像対応
Telegram Bot	低（5 分）	無料	ネイティブアプリで転送	ネイティブ対応
Slack Bot	中	無料/有料	まずまず	対応
Google Forms	低	無料	ブラウザ	アップロードが手間
自作アプリ	高	構成次第	開発が必要	開発が必要

Telegram Bot API は純粋な HTTP API で、SDK のインストールは不要。curl や任意の言語の HTTP クライアントで呼び出せる。Bot の作成に審査もストア登録も不要で、数分でメッセージの受信を開始できる。

Step 1：Telegram Bot を作成して Token を取得する

1. Telegram を開き、@BotFather を検索する（青いチェックマークを確認）
2. /newbot を送信する
3. プロンプトに従って Bot 名（表示名）と username（bot で終わる必要がある。例：my_triage_bot）を入力する
4. BotFather が API Token を返す。形式は 123456:ABC-DEF1234ghIkl-zyx57W2v1u123ew11 のようになる

重要：Token は Bot のパスワードと同義だ。Token を持つ誰もが Bot を完全に制御できる。絶対に git にコミットせず、環境変数で管理すること。

.env ファイルを作成して Token を保存する：

# .env
TELEGRAM_BOT_TOKEN=123456:ABC-DEF1234ghIkl-zyx57W2v1u123ew11

Python で読み込む：

import os
from dotenv import load_dotenv

load_dotenv()
BOT_TOKEN = os.getenv("TELEGRAM_BOT_TOKEN")

Step 2：Chat ID を取得する

Bot がどのチャットを監視するか知る必要がある。最も簡単な方法：

1. Telegram で Bot に任意のメッセージを送信する（例：hello）
2. ブラウザで以下の URL を開く（Token を置き換える）：

https://api.telegram.org/bot<YOUR_TOKEN>/getUpdates

3. 以下のような JSON レスポンスが表示される：

{
  "ok": true,
  "result": [
    {
      "update_id": 123456789,
      "message": {
        "message_id": 1,
        "from": {
          "id": 987654321,
          "first_name": "Alex",
          "is_bot": false
        },
        "chat": {
          "id": 987654321,
          "type": "private"
        },
        "text": "hello"
      }
    }
  ]
}

chat.id が必要な値だ。プライベートチャットは正の数、グループの Chat ID は負の数（例：-1001234567890）になる。

グループでの利用に関する注意

Bot でグループメッセージを監視したい場合、重要なプライバシー設定がある。

Telegram 公式ドキュメントによると、Bot はデフォルトで Privacy Mode が有効になっており、グループでは以下のメッセージのみ表示される：

• その Bot 宛の /command コマンド
• 一般コマンド（Bot がグループで最後にメッセージを送信した Bot の場合）
• その Bot 経由で送信されたインラインメッセージ
• その Bot へのリプライメッセージ

Bot がグループ内のすべてのメッセージを見られるようにするには、2 つの方法がある：

1. Privacy Mode を無効にする：BotFather で Bot Settings → Group Privacy → Turn Off。無効化後は Bot をグループから一度削除して再追加する必要がある。
2. 管理者に設定する：Bot をグループ管理者にすると、すべてのメッセージを受信できる。

Step 3：メッセージの取得：テキスト・画像・送信者

getUpdates API は Telegram Bot の最も重要なメソッドだ。前回のリクエスト以降のすべての新しいメッセージを返す。

主要パラメータ：

パラメータ	型	説明
offset	Integer	どの update_id から返すか。前回の最大 update_id + 1 を設定して重複を防ぐ
limit	Integer	返す件数、1-100、デフォルト 100
timeout	Integer	ロングポーリングの待機秒数、デフォルト 0（即座に返す）。正の整数を設定するとロングポーリングが有効になる

重要な制限：getUpdates は直近 24 時間のデータのみ保持する。 24 時間以上ポーリングしなければ、それ以前のメッセージは永久に失われる。cron job のスケジュール間隔を決める際の最も重要な考慮事項だ。

以下はメッセージを取得する Python 関数：

import requests

def fetch_new_messages(token, offset=None):
    """Telegram から新しいメッセージを取得し、メッセージリストと新しい offset を返す"""
    url = f"https://api.telegram.org/bot{token}/getUpdates"
    params = {"timeout": 10, "limit": 100}
    if offset:
        params["offset"] = offset

    resp = requests.get(url, params=params).json()
    if not resp.get("ok"):
        print(f"Error: {resp}")
        return [], offset

    messages = []
    new_offset = offset

    for update in resp["result"]:
        new_offset = update["update_id"] + 1
        msg = update.get("message", {})
        messages.append({
            "update_id": update["update_id"],
            "chat_id": msg.get("chat", {}).get("id"),
            "from": msg.get("from", {}).get("first_name", "Unknown"),
            "text": msg.get("text"),
            "caption": msg.get("caption"),
            "photo_file_id": msg["photo"][-1]["file_id"] if msg.get("photo") else None,
        })

    return messages, new_offset

ポイント：

• msg["photo"] は配列で、実務上は解像度の低い順に並んでいる（公式ドキュメントには順序の明確な保証はないが、広く認められた動作だ）。最後の要素が最高解像度
• caption は写真に付けられたテキスト説明。スクリーンショットの転送時に説明を追加するユーザーもいる
• offset は継続的に更新する必要がある。さもないと毎回重複メッセージを受信する

Step 4：画像のダウンロード

file_id を取得した後、実際の画像のダウンロードには 2 ステップが必要：

1. getFile API を呼び出して file_path を取得する
2. file_path でダウンロード URL を構成する

import base64

def download_photo(token, file_id):
    """Telegram の写真をダウンロードし、base64 エンコード文字列を返す"""
    # Step 1: file_path を取得
    file_info = requests.get(
        f"https://api.telegram.org/bot{token}/getFile",
        params={"file_id": file_id}
    ).json()

    file_path = file_info["result"]["file_path"]

    # Step 2: 画像をダウンロード
    download_url = f"https://api.telegram.org/file/bot{token}/{file_path}"
    image_data = requests.get(download_url).content

    return base64.b64encode(image_data).decode("utf-8")

注意：Telegram Bot API ドキュメントによると、ダウンロードリンクは最低 1 時間有効で、ファイルサイズの上限は 20MB だ。

Step 5：LLM Vision API によるフィードバック分類

これがシステム全体の核心だ。スクリーンショットをマルチモーダル LLM に送信し、Bug、Feature Request、UX 改善のどれかを自動判定させる。

モデル選択

Vision（画像入力）をサポートする LLM であればこのタスクに対応できる。推奨オプション：

• Google Gemini：無料プランがある唯一の選択肢。1日のリクエスト上限はモデルによって異なる（公式料金ページを参照）。ゼロコストで始めるのに最適
• GPT-4o：エコシステムが成熟しており、ドキュメントや例が豊富で、最も早く始められる
• Claude Haiku 4.5：有料プランの中で最もコストが低く、大量分類に最適

フィードバック分類のような比較的シンプルなタスクであれば、どのモデルでも十分だ。ゼロコストで試したい場合は、まず Gemini API 無料プランをお勧めする。以下の例は GPT-4o をメインにしており、Claude への切り替えは数行の変更で済む（後述）。

OpenAI GPT-4o の例

from openai import OpenAI

client = OpenAI()  # 環境変数 OPENAI_API_KEY から読み込み

SYSTEM_PROMPT = """あなたはプロダクトフィードバック分類アシスタントです。
ユーザーが提供したスクリーンショットとテキスト説明を分析し、
フィードバックの種類を判定して JSON 形式で出力してください。

分類タイプ：
- bug: 明確なエラー、異常動作、クラッシュ
- feature: 新機能リクエストや改善提案
- ux: UI/UX の体験上の問題（バグではないが使いにくい）
- question: 使い方の質問、上記に該当しないもの

出力形式：
{
  "type": "bug|feature|ux|question",
  "priority": "P1|P2|P3",
  "summary": "一行の要約",
  "description": "スクリーンショットから抽出した重要情報を含む詳細説明",
  "components": ["関連する機能モジュール"]
}"""

def classify_feedback(image_base64, text=None):
    """GPT-4o Vision でフィードバックを分類する"""
    content = []

    if text:
        content.append({"type": "text", "text": f"ユーザーの説明：{text}"})

    if image_base64:
        content.append({
            "type": "image_url",
            "image_url": {
                "url": f"data:image/jpeg;base64,{image_base64}",
                "detail": "high"
            }
        })

    response = client.chat.completions.create(
        model="gpt-4o",
        response_format={"type": "json_object"},
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": content}
        ],
        max_tokens=500
    )

    return response.choices[0].message.content

Claude の例

import anthropic

client = anthropic.Anthropic()  # 環境変数 ANTHROPIC_API_KEY から読み込み

def classify_feedback_claude(image_base64, text=None):
    """Claude Vision でフィードバックを分類する"""
    content = []

    if image_base64:
        content.append({
            "type": "image",
            "source": {
                "type": "base64",
                "media_type": "image/jpeg",
                "data": image_base64
            }
        })

    user_text = "このスクリーンショットを分析し、フィードバックを分類してください。"
    if text:
        user_text += f"\nユーザーの説明：{text}"
    content.append({"type": "text", "text": user_text})

    response = client.messages.create(
        model="claude-haiku-4-5-20251001",
        max_tokens=500,
        system=SYSTEM_PROMPT,
        messages=[{"role": "user", "content": content}]
    )

    return response.content[0].text

構造化出力の例

{
  "type": "bug",
  "priority": "P2",
  "summary": "モバイルログインページの SSO フローで 404 エラー",
  "description": "スクリーンショットはモバイル Web で SSO ログイン後にリダイレクト先で 404 エラーページが表示されていることを示している。エラーテキストは 'Service Unavailable'。Google SSO でログインするすべてのモバイルユーザーに影響する。",
  "components": ["Auth", "Mobile Web"]
}

Step 6：完全なポーリングスクリプト

前述のすべての関数を組み合わせた、そのまま実行できる完全なスクリプト：

#!/usr/bin/env python3
"""triage_bot.py - Telegram フィードバック自動分類スクリプト"""

import json
import os
import sys
import base64
import requests
from pathlib import Path
from dotenv import load_dotenv
from openai import OpenAI

load_dotenv()

BOT_TOKEN = os.getenv("TELEGRAM_BOT_TOKEN")
TARGET_CHAT_ID = int(os.getenv("TELEGRAM_CHAT_ID", "0"))
OFFSET_FILE = Path("offset.txt")

openai_client = OpenAI()

# === SYSTEM_PROMPT（Step 5 と同じ）===
SYSTEM_PROMPT = """あなたはプロダクトフィードバック分類アシスタントです。
ユーザーが提供したスクリーンショットとテキスト説明を分析し、
フィードバックの種類を判定して JSON 形式で出力してください。

分類タイプ：
- bug: 明確なエラー、異常動作、クラッシュ
- feature: 新機能リクエストや改善提案
- ux: UI/UX の体験上の問題
- question: 使い方の質問

出力形式：
{"type": "bug|feature|ux|question", "priority": "P1|P2|P3",
 "summary": "一行の要約", "description": "詳細説明",
 "components": ["機能モジュール"]}"""


def load_offset():
    if OFFSET_FILE.exists():
        return int(OFFSET_FILE.read_text().strip())
    return None


def save_offset(offset):
    OFFSET_FILE.write_text(str(offset))


def fetch_new_messages(token, offset=None):
    url = f"https://api.telegram.org/bot{token}/getUpdates"
    params = {"timeout": 10, "limit": 100}
    if offset:
        params["offset"] = offset
    resp = requests.get(url, params=params).json()
    if not resp.get("ok"):
        print(f"[ERROR] getUpdates failed: {resp}", file=sys.stderr)
        return [], offset
    messages, new_offset = [], offset
    for update in resp["result"]:
        new_offset = update["update_id"] + 1
        msg = update.get("message", {})
        if msg.get("chat", {}).get("id") != TARGET_CHAT_ID:
            continue
        messages.append({
            "from": msg.get("from", {}).get("first_name", "Unknown"),
            "text": msg.get("text"),
            "caption": msg.get("caption"),
            "photo_file_id": msg["photo"][-1]["file_id"] if msg.get("photo") else None,
        })
    return messages, new_offset


def download_photo(token, file_id):
    file_info = requests.get(
        f"https://api.telegram.org/bot{token}/getFile",
        params={"file_id": file_id}
    ).json()
    file_path = file_info["result"]["file_path"]
    image_data = requests.get(
        f"https://api.telegram.org/file/bot{token}/{file_path}"
    ).content
    return base64.b64encode(image_data).decode("utf-8")


def classify_feedback(image_base64=None, text=None):
    content = []
    if text:
        content.append({"type": "text", "text": f"ユーザーの説明：{text}"})
    if image_base64:
        content.append({
            "type": "image_url",
            "image_url": {"url": f"data:image/jpeg;base64,{image_base64}", "detail": "high"}
        })
    if not content:
        return None
    response = openai_client.chat.completions.create(
        model="gpt-4o",
        response_format={"type": "json_object"},
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": content}
        ],
        max_tokens=500
    )
    return json.loads(response.choices[0].message.content)


def main():
    offset = load_offset()
    messages, new_offset = fetch_new_messages(BOT_TOKEN, offset)
    save_offset(new_offset)

    if not messages:
        print("[INFO] No new messages.")
        return

    print(f"[INFO] Processing {len(messages)} new message(s)...")

    for msg in messages:
        image_b64 = None
        if msg["photo_file_id"]:
            image_b64 = download_photo(BOT_TOKEN, msg["photo_file_id"])

        text = msg["text"] or msg["caption"]
        result = classify_feedback(image_b64, text)

        if result:
            print(f"\n--- Feedback from {msg['from']} ---")
            print(json.dumps(result, indent=2, ensure_ascii=False))
        else:
            print(f"[SKIP] Empty message from {msg['from']}")


if __name__ == "__main__":
    main()

使い方：

# 依存関係のインストール
pip install requests python-dotenv openai

# 環境変数の設定（.env ファイル）
TELEGRAM_BOT_TOKEN=your_token_here
TELEGRAM_CHAT_ID=your_chat_id_here
OPENAI_API_KEY=your_openai_key_here

# 実行
python triage_bot.py

スクリプトは offset を offset.txt に保存し、毎回前回以降の新しいメッセージのみを処理する。

Step 7：Cron Job による自動スケジューリング

手動実行はテストに適しているが、実運用では自動スケジューリングが必要だ。

Webhook ではなく Cron を使う理由

比較	Cron + ポーリング	Webhook
公開サーバーの要否	不要	必要（HTTPS + 有効な SSL）
設定難易度	低（crontab 1 行）	中（DNS + SSL + サーバー）
リアルタイム性	cron 間隔に依存	リアルタイム
適する場面	個人/小規模チーム	高トラフィックの本番環境

注意：ポーリング（getUpdates）と Webhook は排他的だ。Webhook を設定すると、getUpdates は正常に動作しなくなる。ポーリングに戻すには、まず deleteWebhook API を呼び出す必要がある。

Crontab の設定

# crontab エディタを開く
crontab -e

# 2 時間ごとに実行（推奨間隔、24 時間の保持制限よりも十分短い）
0 */2 * * * cd /path/to/your/project && /usr/bin/python3 triage_bot.py >> triage.log 2>&1

cron 間隔の重要な考慮事項：getUpdates は 24 時間分のデータしか保持しないため、cron 間隔は 24 時間よりも十分短くする必要がある。2〜4 時間の間隔を推奨する。1 回の実行が失敗しても、次の実行でデータが期限切れになる前に回収できるバッファを確保しよう。

macOS ユーザー向け

macOS では権限の問題で cron が不安定になることがある。代わりに launchd を使用する。以下の plist を ~/Library/LaunchAgents/com.triage-bot.plist として保存する：

<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN"
  "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
<plist version="1.0">
<dict>
  <key>Label</key>
  <string>com.triage-bot</string>
  <key>ProgramArguments</key>
  <array>
    <string>/usr/bin/python3</string>
    <string>/path/to/your/project/triage_bot.py</string>
  </array>
  <key>WorkingDirectory</key>
  <string>/path/to/your/project</string>
  <key>StartInterval</key>
  <integer>7200</integer>
  <key>StandardOutPath</key>
  <string>/path/to/your/project/triage.log</string>
  <key>StandardErrorPath</key>
  <string>/path/to/your/project/triage-error.log</string>
</dict>
</plist>

スケジュールの読み込み：

launchctl load ~/Library/LaunchAgents/com.triage-bot.plist

ローコード代替案：n8n

ビジュアルなワークフローを好む場合、n8n にはビルトインの Telegram Trigger ノードがあり、ドラッグ＆ドロップで設定できる。コードは不要だ。

応用例

このシステムの核心は「メッセージ取得 + AI 分類」だ。分類結果が出れば、さまざまな下流アクションに接続できる：

• Jira / Linear チケットの自動作成：REST API を呼び出して分類結果をそのままチケットに変換
• 分類結果を Telegram に返信：sendMessage API で Bot がグループ内に分類サマリーを返信し、チームがリアルタイムで確認
• Google Sheets / Notion に保存：API で構造化データを書き込み、検索可能なフィードバック履歴を構築
• Webhook モードの追加：公開サーバーがある場合、Webhook に切り替えてリアルタイム処理を実現

リスクと制限事項

Token のセキュリティ

Bot Token は Bot の完全な制御権を与える。環境変数または secrets manager に保存し、絶対にコードにハードコードしないこと。漏洩が疑われる場合は、すぐに BotFather で /revoke を使って再生成する。

24 時間のデータ保持

getUpdates のデータは 24 時間しか保持されない。cron job が何らかの理由で 1 日以上停止した場合（サーバーの再起動、パスエラー、権限の問題）、その期間のフィードバックは永久に失われる。簡単なヘルスチェック機構の導入を推奨する。例えば実行ごとにタイムスタンプを記録し、外部モニタリングでタイムスタンプの鮮度を確認する。

LLM 分類の精度

Vision API の分類は 100% 正確ではない。実際の使用経験では、明確なバグのスクリーンショット（エラーメッセージ付き）は高い精度で分類されるが、曖昧な UX 問題や Feature Request の境界では AI の判断が期待と異なることがある。優先度の高いフィードバックは人的レビューを推奨する。

スケール時のコスト

低用量（1日 10〜20 枚のスクリーンショット）はほぼ無料だ。しかし画像ごとに API コールが必要なため、フィードバック量が増えるとコストは線形に増加する。無料枠または最低プランから始めて、実際の使用量を観察してからモデルを決定するのがよい。各社の Vision API 料金は OpenAI Pricing と Anthropic Pricing を参照。

プライバシーへの配慮

ユーザーのスクリーンショットには個人情報（氏名、メールアドレス、パスワードさえ）が含まれている可能性がある。これらの画像はサードパーティの LLM API に送信されて分析される。チームに厳格なデータプライバシー要件がある場合、これがデータ処理ポリシーに適合するか評価が必要だ。

Telegram Rate Limits

Telegram 公式 FAQ によると、Rate Limit は階層構造になっている：グローバルで約 30 msg/sec（異なるチャットへ）、同一チャットで約 1 msg/sec、同一グループで 20 msg/min。getUpdates のポーリングでは通常これらの制限に達することはないが、Bot が大量のメッセージを返信する場合は注意が必要だ。

まとめ

このシステムの魅力はシンプルさにある：1 つの Python ファイル、1 行の crontab、1 つの API キー。フレームワークもデプロイも常時稼働サーバーも不要だ。

まず Step 1〜3 から始めよう：Bot を作成し、Chat ID を取得し、メッセージが取得できることを確認する。この 3 ステップが完了すれば、動作するメッセージ取得パイプラインが手に入る。その後 LLM 分類と cron job を追加し、最後に Jira や普段使っているツールに接続しよう。

「スクリーンショットを転送するだけで自動分類される」便利さを体験すれば、他にどんなワークフローを接続できるか考え始めるはずだ。