Running a Gemini + LINE Bot in Production — Reply Token Expiry, Duplicate Replies, and Cold-Start Latency

# app.py
import os
import threading
from flask import Flask, request, abort
from google import genai
from linebot.v3 import WebhookParser
from linebot.v3.messaging import (
    Configuration, ApiClient, MessagingApi,
    ReplyMessageRequest, PushMessageRequest, TextMessage,
    ShowLoadingAnimationRequest,
)
from linebot.v3.webhooks import MessageEvent, TextMessageContent
 
app = Flask(__name__)
 
GEMINI_API_KEY = os.environ["GEMINI_API_KEY"]
LINE_CHANNEL_SECRET = os.environ["LINE_CHANNEL_SECRET"]
LINE_CHANNEL_ACCESS_TOKEN = os.environ["LINE_CHANNEL_ACCESS_TOKEN"]
 
gemini_client = genai.Client(api_key=GEMINI_API_KEY)
configuration = Configuration(access_token=LINE_CHANNEL_ACCESS_TOKEN)
parser = WebhookParser(LINE_CHANNEL_SECRET)
 
 
def generate_reply(user_id: str, user_message: str) -> str:
    """Generate a response with Gemini, using conversation history."""
    history = load_history(user_id)  # from Firestore (below)
    history.append({"role": "user", "parts": [{"text": user_message}]})
 
    response = gemini_client.models.generate_content(
        model="gemini-3-flash",
        contents=history,
        config={
            "system_instruction": (
                "You are a friendly assistant running on LINE. "
                "Answer concisely, around 200 characters."
            ),
            "temperature": 0.7,
            "max_output_tokens": 500,
        },
    )
    answer = response.text
    history.append({"role": "model", "parts": [{"text": answer}]})
    save_history(user_id, history)  # to Firestore (below)
    return answer
 
 
def process_event(user_id: str, reply_token: str, text: str):
    """Runs on a separate thread: generate, then reply or push."""
    with ApiClient(configuration) as api_client:
        api = MessagingApi(api_client)
        # Show a typing indicator so the user knows we're thinking
        api.show_loading_animation(
            ShowLoadingAnimationRequest(chat_id=user_id, loading_seconds=20)
        )
        answer = generate_reply(user_id, text)
        try:
            api.reply_message(
                ReplyMessageRequest(
                    reply_token=reply_token,
                    messages=[TextMessage(text=answer)],
                )
            )
        except Exception:
            # Token expired — fall back to push
            api.push_message(
                PushMessageRequest(
                    to=user_id, messages=[TextMessage(text=answer)]
                )
            )
 
 
@app.route("/callback", methods=["POST"])
def callback():
    signature = request.headers.get("X-Line-Signature", "")
    body = request.get_data(as_text=True)
    try:
        events = parser.parse(body, signature)
    except Exception:
        abort(400)
 
    for event in events:
        if isinstance(event, MessageEvent) and isinstance(
            event.message, TextMessageContent
        ):
            if already_handled(event.webhook_event_id):  # idempotency (below)
                continue
            threading.Thread(
                target=process_event,
                args=(event.source.user_id, event.reply_token, event.message.text),
                daemon=True,
            ).start()
 
    return "OK"  # return 200 without waiting for generation
 
 
@app.route("/health")
def health():
    return "OK"

from google.cloud import firestore
 
db = firestore.Client()
 
def already_handled(event_id: str) -> bool:
    """Record the event ID; return False only on the first sight."""
    ref = db.collection("handled_events").document(event_id)
    # create() raises if the document already exists — use that as a lock
    try:
        ref.create({"ts": firestore.SERVER_TIMESTAMP})
        return False  # first time
    except Exception:
        return True   # redelivery / duplicate

MAX_TURNS = 20  # last 20 messages (about 10 user+model exchanges)
 
def load_history(user_id: str) -> list[dict]:
    doc = db.collection("conversations").document(user_id).get()
    return doc.to_dict().get("history", []) if doc.exists else []
 
def save_history(user_id: str, history: list[dict]):
    trimmed = history[-MAX_TURNS:]
    db.collection("conversations").document(user_id).set({"history": trimmed})

# Terminal 1
export GEMINI_API_KEY="your_key"
export LINE_CHANNEL_SECRET="your_secret"
export LINE_CHANNEL_ACCESS_TOKEN="your_token"
python app.py
 
# Terminal 2
ngrok http 8080
# https://xxxx-xxx-xxx.ngrok-free.app

import requests, base64
from linebot.v3.webhooks import ImageMessageContent
 
def download_line_image(message_id: str) -> bytes:
    url = f"https://api-data.line.me/v2/bot/message/{message_id}/content"
    headers = {"Authorization": f"Bearer {LINE_CHANNEL_ACCESS_TOKEN}"}
    return requests.get(url, headers=headers).content
 
def describe_image(image_bytes: bytes) -> str:
    image_b64 = base64.b64encode(image_bytes).decode("utf-8")
    response = gemini_client.models.generate_content(
        model="gemini-3-flash",
        contents=[{
            "role": "user",
            "parts": [
                {"text": "Describe this image briefly."},
                {"inline_data": {"mime_type": "image/jpeg", "data": image_b64}},
            ],
        }],
        config={"max_output_tokens": 500},
    )
    return response.text
 
# Add to the callback branch:
# elif isinstance(event.message, ImageMessageContent):
#     handle it on a thread like process_event, choosing reply vs push

FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY app.py .
CMD ["gunicorn", "--bind", "0.0.0.0:8080", "--workers", "2", "--threads", "8", "app:app"]

gcloud run deploy gemini-line-bot \
  --source . \
  --region asia-northeast1 \
  --allow-unauthenticated \
  --set-env-vars "GEMINI_API_KEY=...,LINE_CHANNEL_SECRET=...,LINE_CHANNEL_ACCESS_TOKEN=..." \
  --memory 512Mi \
  --min-instances 0 \
  --max-instances 10

import time
from google.api_core.exceptions import ResourceExhausted, ServiceUnavailable
 
def generate_with_retry(contents, config, max_retries=3):
    for attempt in range(max_retries):
        try:
            return gemini_client.models.generate_content(
                model="gemini-3-flash", contents=contents, config=config,
            )
        except (ResourceExhausted, ServiceUnavailable):
            wait = 2 ** attempt  # 1s, 2s, 4s
            time.sleep(wait)
    raise RuntimeError("Gemini retries exhausted")