Gemini API × Slack Bot 本番実装 — Bolt SDK・スレッドコンテキスト・Cloud Run デプロイまで

# app/gemini_client.py
import asyncio
import time
import logging
from typing import Optional
from google import genai
from google.genai import types
 
logger = logging.getLogger(__name__)
 
class GeminiClient:
    """
    Gemini API ラッパー。レート制限・リトライ処理を内包する。
    """
 
    def __init__(self, api_key: str, model: str = "gemini-2.5-flash"):
        self.client = genai.Client(api_key=api_key)
        self.model = model
        self._request_count = 0
        self._window_start = time.time()
 
    async def generate(
        self,
        contents: list,
        system_instruction: Optional[str] = None,
        max_retries: int = 3,
    ) -> str:
        """
        コンテンツリストを受け取りテキスト応答を生成する。
        429エラー（レート制限）は指数バックオフでリトライする。
        """
        config = types.GenerateContentConfig(
            system_instruction=system_instruction or self._default_system_instruction(),
            max_output_tokens=4096,
            temperature=0.7,
        )
 
        for attempt in range(max_retries):
            try:
                response = await asyncio.to_thread(
                    self.client.models.generate_content,
                    model=self.model,
                    contents=contents,
                    config=config,
                )
                self._request_count += 1
                return response.text or ""
 
            except Exception as e:
                error_str = str(e)
                if "429" in error_str or "RESOURCE_EXHAUSTED" in error_str:
                    wait = 2 ** attempt * 5  # 5秒, 10秒, 20秒
                    logger.warning(
                        f"Rate limited (attempt {attempt+1}/{max_retries}). "
                        f"Waiting {wait}s..."
                    )
                    await asyncio.sleep(wait)
                    if attempt == max_retries - 1:
                        return "⚠️ ただいまリクエストが集中しています。しばらくしてからもう一度お試しください。"
                else:
                    logger.error(f"Gemini API error: {e}")
                    raise
 
        return "⚠️ 応答の生成中にエラーが発生しました。"
 
    def _default_system_instruction(self) -> str:
        return (
            "あなたは Slack で動作する AI アシスタントです。"
            "チームのメンバーの質問に対して、丁寧かつ正確に回答してください。"
            "Slack の Markdown 記法（`*太字*`、`_斜体_`、` ``` コードブロック ``` `）を活用して"
            "見やすいフォーマットで回答してください。"
            "回答は簡潔にまとめ、必要に応じてリストや番号付きリストを使ってください。"
        )

# app/gemini_client.py への追加メソッド
 
    async def generate_with_image(
        self,
        text: str,
        image_bytes: bytes,
        mime_type: str = "image/png",
        history: Optional[list] = None,
    ) -> str:
        """
        画像つきプロンプトで Gemini に問い合わせる。
        Slack から共有された画像を解析するために使用する。
        """
        # 画像パーツを構築
        image_part = types.Part.from_bytes(data=image_bytes, mime_type=mime_type)
        text_part = types.Part.from_text(text=text)
 
        contents = []
 
        # 会話履歴を先頭に追加
        if history:
            contents.extend(history)
 
        # 最新メッセージ（画像 + テキスト）を追加
        contents.append(types.Content(
            role="user",
            parts=[image_part, text_part],
        ))
 
        return await self.generate(contents=contents)

# app/context_store.py
import time
import logging
from collections import defaultdict
from typing import Optional
from google.genai import types
 
logger = logging.getLogger(__name__)
 
# 会話履歴の保持上限
MAX_HISTORY_TURNS = 10   # 10ターン（ユーザー発言 + Bot 応答 = 1ターン）
CONTEXT_TTL_SEC = 3600   # 1時間でコンテキストを破棄
 
class ThreadContextStore:
    """
    Slack スレッド ID をキーとして Gemini 用の会話履歴を管理するストア。
 
    本番環境では Redis / Firestore に差し替えることを想定している。
    """
 
    def __init__(self):
        # { thread_ts: {"history": [...], "last_accessed": timestamp} }
        self._store: dict = defaultdict(lambda: {"history": [], "last_accessed": time.time()})
 
    def get_history(self, thread_ts: str) -> list:
        """指定スレッドの会話履歴を返す。TTL 切れは空リスト。"""
        entry = self._store.get(thread_ts)
        if not entry:
            return []
 
        if time.time() - entry["last_accessed"] > CONTEXT_TTL_SEC:
            # TTL 切れ → コンテキストをリセット
            logger.info(f"Context TTL expired for thread {thread_ts}")
            del self._store[thread_ts]
            return []
 
        return entry["history"]
 
    def add_turn(self, thread_ts: str, user_text: str, bot_text: str):
        """ユーザー発言と Bot 応答を1ターンとして履歴に追加する。"""
        entry = self._store[thread_ts]
        entry["last_accessed"] = time.time()
 
        # Gemini API 形式の Content オブジェクトで保存
        entry["history"].append(
            types.Content(role="user", parts=[types.Part.from_text(text=user_text)])
        )
        entry["history"].append(
            types.Content(role="model", parts=[types.Part.from_text(text=bot_text)])
        )
 
        # 上限を超えた場合は古いターンから削除（2件 = 1ターン）
        while len(entry["history"]) > MAX_HISTORY_TURNS * 2:
            entry["history"].pop(0)
            entry["history"].pop(0)
 
    def clear(self, thread_ts: str):
        """スレッドのコンテキストをリセットする。"""
        if thread_ts in self._store:
            del self._store[thread_ts]
            logger.info(f"Context cleared for thread {thread_ts}")

# app/bot.py
import logging
import httpx
from slack_bolt.async_app import AsyncApp
from slack_bolt.adapter.fastapi.async_handler import AsyncSlackRequestHandler
from app.gemini_client import GeminiClient
from app.context_store import ThreadContextStore
from app.config import Settings
 
logger = logging.getLogger(__name__)
 
settings = Settings()
 
# Bolt アプリの初期化（Socket Mode または HTTP モード）
app = AsyncApp(
    token=settings.slack_bot_token,
    signing_secret=settings.slack_signing_secret,
)
 
gemini = GeminiClient(api_key=settings.gemini_api_key)
context_store = ThreadContextStore()
handler = AsyncSlackRequestHandler(app)
 
def _get_thread_ts(event: dict) -> str:
    """スレッドのルート TS を返す。スレッド内なら thread_ts、単発なら ts。"""
    return event.get("thread_ts") or event.get("ts", "")
 
@app.event("app_mention")
async def handle_mention(event: dict, say, client):
    """
    Bot がメンションされたときの処理。
    スレッド内であれば、そのスレッドの会話コンテキストを引き継ぐ。
    """
    thread_ts = _get_thread_ts(event)
    channel_id = event["channel"]
 
    # Bot のメンション部分を除去したユーザーメッセージ
    user_message = _strip_mention(event.get("text", ""))
 
    if not user_message.strip():
        await say(
            text="こんにちは！何かお手伝いできることはありますか？",
            thread_ts=thread_ts,
        )
        return
 
    # 「クリア」コマンドでコンテキストをリセット
    if user_message.strip().lower() in ["clear", "クリア", "リセット"]:
        context_store.clear(thread_ts)
        await say(text="✅ 会話履歴をリセットしました。", thread_ts=thread_ts)
        return
 
    # 画像添付の確認
    files = event.get("files", [])
    image_file = next(
        (f for f in files if f.get("mimetype", "").startswith("image/")),
        None,
    )
 
    # タイピングインジケーターを送信
    await client.chat_postMessage(
        channel=channel_id,
        thread_ts=thread_ts,
        text="⏳ 考え中...",
    )
 
    try:
        history = context_store.get_history(thread_ts)
 
        if image_file:
            # 画像 + テキストのマルチモーダル処理
            image_bytes = await _download_slack_file(
                image_file["url_private"],
                settings.slack_bot_token,
            )
            response_text = await gemini.generate_with_image(
                text=user_message or "この画像について説明してください。",
                image_bytes=image_bytes,
                mime_type=image_file["mimetype"],
                history=history,
            )
        else:
            # テキストのみの処理
            from google.genai import types
            contents = list(history) + [
                types.Content(
                    role="user",
                    parts=[types.Part.from_text(text=user_message)],
                )
            ]
            response_text = await gemini.generate(contents=contents)
 
        # コンテキストに追加
        context_store.add_turn(thread_ts, user_message, response_text)
 
        await say(text=response_text, thread_ts=thread_ts)
 
    except Exception as e:
        logger.error(f"Error processing message: {e}", exc_info=True)
        await say(
            text="⚠️ エラーが発生しました。しばらく経ってからもう一度お試しください。",
            thread_ts=thread_ts,
        )
 
@app.event("message")
async def handle_dm(event: dict, say):
    """
    DM（Direct Message）のハンドリング。
    DM ではメンション不要で直接 Bot に話しかけられる。
    """
    # Bot 自身のメッセージは無視
    if event.get("bot_id") or event.get("subtype"):
        return
 
    thread_ts = _get_thread_ts(event)
    user_message = event.get("text", "")
 
    if not user_message.strip():
        return
 
    try:
        from google.genai import types
        history = context_store.get_history(thread_ts)
        contents = list(history) + [
            types.Content(role="user", parts=[types.Part.from_text(text=user_message)])
        ]
        response_text = await gemini.generate(contents=contents)
        context_store.add_turn(thread_ts, user_message, response_text)
        await say(text=response_text, thread_ts=thread_ts)
 
    except Exception as e:
        logger.error(f"DM error: {e}", exc_info=True)
        await say(text="⚠️ エラーが発生しました。", thread_ts=thread_ts)
 
def _strip_mention(text: str) -> str:
    """<@BOTID> 形式のメンションをテキストから除去する。"""
    import re
    return re.sub(r"<@[A-Z0-9]+>", "", text).strip()
 
async def _download_slack_file(url: str, token: str) -> bytes:
    """Slack のプライベートファイルをダウンロードして bytes を返す。"""
    async with httpx.AsyncClient() as client:
        response = await client.get(
            url,
            headers={"Authorization": f"Bearer {token}"},
            follow_redirects=True,
        )
        response.raise_for_status()
        return response.content

# app/config.py
import os
from functools import lru_cache
from pydantic_settings import BaseSettings
 
class Settings(BaseSettings):
    """
    環境変数または Secret Manager から設定を読み込む。
    ローカル開発では .env ファイルを利用する。
    """
    slack_bot_token: str = ""
    slack_signing_secret: str = ""
    slack_app_token: str = ""  # Socket Mode 用（開発環境のみ）
    gemini_api_key: str = ""
    environment: str = "development"  # "development" | "production"
    gcp_project_id: str = ""
 
    class Config:
        env_file = ".env"
 
    def load_from_secret_manager(self):
        """
        Cloud Run 本番環境では Secret Manager から認証情報を取得する。
        環境変数 ENVIRONMENT=production のときのみ実行。
        """
        if self.environment != "production":
            return
 
        from google.cloud import secretmanager
        client = secretmanager.SecretManagerServiceClient()
 
        def _get_secret(name: str) -> str:
            path = f"projects/{self.gcp_project_id}/secrets/{name}/versions/latest"
            response = client.access_secret_version(request={"name": path})
            return response.payload.data.decode("UTF-8")
 
        self.slack_bot_token = _get_secret("slack-bot-token")
        self.slack_signing_secret = _get_secret("slack-signing-secret")
        self.gemini_api_key = _get_secret("gemini-api-key")
 
@lru_cache()
def get_settings() -> Settings:
    s = Settings()
    s.load_from_secret_manager()
    return s

# app/main.py
import logging
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from app.bot import handler
 
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
 
api = FastAPI(title="Gemini Slack Bot")
 
@api.post("/slack/events")
async def slack_events(req: Request):
    """Slack Events API のエンドポイント。"""
    return await handler.handle(req)
 
@api.get("/health")
async def health():
    """Cloud Run のヘルスチェック用エンドポイント。"""
    return JSONResponse({"status": "ok"})

# app/bot.py への追加
 
@app.command("/gemini-help")
async def help_command(ack, respond):
    """ヘルプメッセージを表示する Slash Command。"""
    await ack()
    await respond(
        text=(
            "*🤖 Gemini Bot の使い方*\n\n"
            "• `@GeminiBot [質問]` — 何でも質問できます\n"
            "• `@GeminiBot [質問] + 画像添付` — 画像を分析します\n"
            "• `@GeminiBot クリア` — 会話履歴をリセットします\n"
            "• `/gemini-help` — このヘルプを表示します\n\n"
            "スレッド内でメンションすると、会話のコンテキストを引き継ぎます。"
        )
    )
 
@app.command("/gemini-summarize")
async def summarize_command(ack, respond, command):
    """
    入力したテキストを要約する Slash Command。
    使い方: /gemini-summarize [長いテキスト]
    """
    await ack()
    text = command.get("text", "").strip()
 
    if not text:
        await respond(text="使い方: `/gemini-summarize [要約したいテキスト]`")
        return
 
    from google.genai import types
    from app.bot import gemini
 
    contents = [types.Content(
        role="user",
        parts=[types.Part.from_text(
            text=f"以下のテキストを3〜5行の日本語で簡潔に要約してください:\n\n{text}"
        )],
    )]
    summary = await gemini.generate(contents=contents)
    await respond(text=f"*📝 要約:*\n{summary}")

# Dockerfile
FROM python:3.11-slim
 
WORKDIR /app
 
# 依存関係のインストール
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
 
# アプリケーションコードのコピー
COPY app/ ./app/
 
# Cloud Run のデフォルトポート
ENV PORT=8080
EXPOSE 8080
 
CMD ["uvicorn", "app.main:api", "--host", "0.0.0.0", "--port", "8080", "--workers", "2"]

# シークレットの登録（初回のみ）
gcloud secrets create slack-bot-token --replication-policy="automatic"
echo -n "xoxb-your-slack-bot-token" | \
  gcloud secrets versions add slack-bot-token --data-file=-
 
gcloud secrets create slack-signing-secret --replication-policy="automatic"
echo -n "your-signing-secret" | \
  gcloud secrets versions add slack-signing-secret --data-file=-
 
gcloud secrets create gemini-api-key --replication-policy="automatic"
echo -n "YOUR_GEMINI_API_KEY" | \
  gcloud secrets versions add gemini-api-key --data-file=-

# プロジェクト設定
PROJECT_ID="your-gcp-project-id"
REGION="asia-northeast1"  # 東京リージョン
SERVICE_NAME="gemini-slack-bot"
 
# Docker イメージのビルド & プッシュ
gcloud builds submit --tag "gcr.io/${PROJECT_ID}/${SERVICE_NAME}"
 
# Cloud Run へのデプロイ
gcloud run deploy "${SERVICE_NAME}" \
  --image "gcr.io/${PROJECT_ID}/${SERVICE_NAME}" \
  --platform managed \
  --region "${REGION}" \
  --allow-unauthenticated \
  --memory 512Mi \
  --concurrency 80 \
  --timeout 60s \
  --set-env-vars "ENVIRONMENT=production,GCP_PROJECT_ID=${PROJECT_ID}" \
  --service-account "gemini-slack-bot@${PROJECT_ID}.iam.gserviceaccount.com"
 
# サービスアカウントに Secret Manager のアクセス権を付与
gcloud projects add-iam-policy-binding "${PROJECT_ID}" \
  --member "serviceAccount:gemini-slack-bot@${PROJECT_ID}.iam.gserviceaccount.com" \
  --role "roles/secretmanager.secretAccessor"

# app/main.py への追加（本番環境のみ）
import os
 
if os.getenv("ENVIRONMENT") == "production":
    import google.cloud.logging
    client = google.cloud.logging.Client()
    client.setup_logging()

# app/rate_limiter.py
import asyncio
import time
from collections import deque
 
class TokenBucketLimiter:
    """
    トークンバケット方式のレート制限器。
    Gemini API の RPM 制限に合わせて調整する。
    """
    def __init__(self, rpm: int = 60):
        self.rpm = rpm
        self.tokens = rpm
        self.last_refill = time.time()
        self._lock = asyncio.Lock()
 
    async def acquire(self):
        async with self._lock:
            now = time.time()
            elapsed = now - self.last_refill
 
            # 経過時間に応じてトークンを補充（最大 rpm まで）
            refill = elapsed * (self.rpm / 60.0)
            self.tokens = min(self.rpm, self.tokens + refill)
            self.last_refill = now
 
            if self.tokens >= 1:
                self.tokens -= 1
                return True
 
            # トークン不足 → 待機時間を計算
            wait_time = (1 - self.tokens) / (self.rpm / 60.0)
            await asyncio.sleep(wait_time)
            self.tokens = 0
            return True

pip install slack-bolt google-genai python-dotenv

# .env
SLACK_BOT_TOKEN=xoxb-your-bot-token
SLACK_APP_TOKEN=xapp-your-app-token
SLACK_SIGNING_SECRET=your-signing-secret
GEMINI_API_KEY=your-gemini-api-key

import os
from dotenv import load_dotenv
from slack_bolt import App
from slack_bolt.adapter.socket_mode import SocketModeHandler
from google import genai
 
load_dotenv()
 
# Slack Appの初期化
app = App(token=os.environ["SLACK_BOT_TOKEN"])
 
# Gemini クライアントの初期化
client = genai.Client(api_key=os.environ["GEMINI_API_KEY"])
 
SYSTEM_INSTRUCTION = """あなたはチームをサポートするAIアシスタントです。
質問に対して正確かつ簡潔に回答してください。
コードの質問にはコード例を含めてください。
日本語で質問された場合は日本語で、英語の場合は英語で回答してください。"""
 
@app.event("app_mention")
def handle_mention(event, say):
    """Botへのメンションに応答する"""
    user_message = event["text"]
    thread_ts = event.get("thread_ts", event["ts"])
 
    try:
        # Gemini APIにリクエスト
        response = client.models.generate_content(
            model="gemini-2.5-flash",
            contents=user_message,
            config=genai.types.GenerateContentConfig(
                system_instruction=SYSTEM_INSTRUCTION,
                max_output_tokens=2048,
                temperature=0.7,
            ),
        )
 
        # Slackに回答を投稿（スレッドに返信）
        say(text=response.text, thread_ts=thread_ts)
 
    except Exception as e:
        say(
            text=f"申し訳ありません、エラーが発生しました: {str(e)}",
            thread_ts=thread_ts,
        )
 
if __name__ == "__main__":
    handler = SocketModeHandler(app, os.environ["SLACK_APP_TOKEN"])
    print("⚡ Gemini Slack Bot が起動しました")
    handler.start()

from google.genai import types
 
def get_thread_history(client_slack, channel, thread_ts):
    """スレッドの会話履歴を取得してGemini形式に変換する"""
    result = client_slack.conversations_replies(
        channel=channel, ts=thread_ts, limit=20
    )
 
    contents = []
    for msg in result["messages"]:
        # Botのメッセージかユーザーのメッセージかを判定
        if msg.get("bot_id"):
            role = "model"
        else:
            role = "user"
 
        contents.append(
            types.Content(
                role=role,
                parts=[types.Part(text=msg["text"])],
            )
        )
 
    return contents
 
@app.event("app_mention")
def handle_mention_with_context(event, say, client):
    """スレッドのコンテキストを維持してAIが応答する"""
    channel = event["channel"]
    thread_ts = event.get("thread_ts", event["ts"])
 
    # スレッド履歴を取得
    contents = get_thread_history(client, channel, thread_ts)
 
    try:
        gemini_client = genai.Client(api_key=os.environ["GEMINI_API_KEY"])
        response = gemini_client.models.generate_content(
            model="gemini-2.5-flash",
            contents=contents,
            config=genai.types.GenerateContentConfig(
                system_instruction=SYSTEM_INSTRUCTION,
                max_output_tokens=2048,
                temperature=0.7,
            ),
        )
 
        say(text=response.text, thread_ts=thread_ts)
 
    except Exception as e:
        say(
            text=f"エラーが発生しました: {str(e)}",
            thread_ts=thread_ts,
        )

from google.genai import types
 
# ツール定義
weather_tool = types.Tool(
    function_declarations=[
        types.FunctionDeclaration(
            name="get_weather",
            description="指定した都市の現在の天気情報を取得します",
            parameters=types.Schema(
                type="OBJECT",
                properties={
                    "city": types.Schema(
                        type="STRING",
                        description="天気を取得する都市名（例: 東京、大阪）",
                    ),
                },
                required=["city"],
            ),
        ),
        types.FunctionDeclaration(
            name="create_task",
            description="タスク管理システムに新しいタスクを作成します",
            parameters=types.Schema(
                type="OBJECT",
                properties={
                    "title": types.Schema(
                        type="STRING", description="タスクのタイトル"
                    ),
                    "assignee": types.Schema(
                        type="STRING", description="担当者の名前"
                    ),
                    "priority": types.Schema(
                        type="STRING",
                        description="優先度",
                        enum=["high", "medium", "low"],
                    ),
                },
                required=["title"],
            ),
        ),
    ]
)
 
def execute_function(function_call):
    """Function Callの結果を実行して返す"""
    name = function_call.name
    args = function_call.args
 
    if name == "get_weather":
        # 実際のAPIに置き換えてください
        return {"city": args["city"], "temp": "22°C", "condition": "晴れ"}
    elif name == "create_task":
        # 実際のタスク管理APIに置き換えてください
        return {"status": "created", "title": args["title"]}
    else:
        return {"error": f"Unknown function: {name}"}
 
@app.event("app_mention")
def handle_with_tools(event, say, client):
    """Function Calling対応のメンションハンドラ"""
    channel = event["channel"]
    thread_ts = event.get("thread_ts", event["ts"])
    user_message = event["text"]
 
    gemini_client = genai.Client(api_key=os.environ["GEMINI_API_KEY"])
 
    try:
        response = gemini_client.models.generate_content(
            model="gemini-2.5-flash",
            contents=user_message,
            config=genai.types.GenerateContentConfig(
                system_instruction=SYSTEM_INSTRUCTION,
                tools=[weather_tool],
                temperature=0.7,
            ),
        )
 
        # Function Callが返された場合の処理
        if response.candidates[0].content.parts[0].function_call:
            fc = response.candidates[0].content.parts[0].function_call
            result = execute_function(fc)
 
            # 結果をGeminiに渡して最終回答を生成
            response = gemini_client.models.generate_content(
                model="gemini-2.5-flash",
                contents=[
                    types.Content(
                        role="user", parts=[types.Part(text=user_message)]
                    ),
                    response.candidates[0].content,
                    types.Content(
                        role="user",
                        parts=[
                            types.Part(
                                function_response=types.FunctionResponse(
                                    name=fc.name, response=result
                                )
                            )
                        ],
                    ),
                ],
                config=genai.types.GenerateContentConfig(
                    system_instruction=SYSTEM_INSTRUCTION,
                    tools=[weather_tool],
                ),
            )
 
        say(text=response.text, thread_ts=thread_ts)
 
    except Exception as e:
        say(text=f"エラーが発生しました: {str(e)}", thread_ts=thread_ts)

import time
import logging
from functools import wraps
 
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
 
def retry_with_backoff(max_retries=3, base_delay=1.0):
    """指数バックオフ付きリトライデコレータ"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_retries - 1:
                        raise
                    delay = base_delay * (2 ** attempt)
                    logger.warning(
                        f"Attempt {attempt + 1} failed: {e}. "
                        f"Retrying in {delay}s..."
                    )
                    time.sleep(delay)
        return wrapper
    return decorator
 
@retry_with_backoff(max_retries=3)
def call_gemini(contents, tools=None):
    """リトライ付きのGemini API呼び出し"""
    gemini_client = genai.Client(api_key=os.environ["GEMINI_API_KEY"])
    config = genai.types.GenerateContentConfig(
        system_instruction=SYSTEM_INSTRUCTION,
        max_output_tokens=2048,
        temperature=0.7,
    )
    if tools:
        config.tools = tools
 
    return gemini_client.models.generate_content(
        model="gemini-2.5-flash",
        contents=contents,
        config=config,
    )

def split_message(text, max_length=3000):
    """長いメッセージを分割する"""
    if len(text) <= max_length:
        return [text]
 
    chunks = []
    while text:
        if len(text) <= max_length:
            chunks.append(text)
            break
        # 改行位置で分割
        split_pos = text.rfind("\n", 0, max_length)
        if split_pos == -1:
            split_pos = max_length
        chunks.append(text[:split_pos])
        text = text[split_pos:].lstrip()
 
    return chunks

FROM python:3.12-slim
 
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
 
COPY . .
CMD ["python", "bot.py"]