Gemini + Cloud Run functions + Firestore：构建记住对话上下文的 LINE AI 助手

把 Gemini 接进日常使用的聊天工具，是理解大模型应用链路的一条捷径。本教程要搭的是一个跑在 LINE 里的私人 AI 助手：用户在 LINE 里发消息，后端函数接收 Webhook，从数据库里取出这个用户之前的对话记录，连同新消息一起交给 Gemini 生成回复，再把回复推回 LINE，并把这一轮问答写回数据库。整条链路走完，你会亲手实现「接收输入 → 保持上下文 → 调用模型 → 返回结果」这个 LLM 应用的基本闭环，也能看清网页版聊天应用背后大致发生了什么。

这套结构适合已经会一点 Python、想动手跑通一次完整链路的人。它依赖 Google Cloud 的免费额度，个人自用场景下运行成本可以压得很低。如果你只是想先单独试试 Gemini 的对话能力，也可以先用站内的 Gemini 工具熟悉一下交互方式，再来搭这套带持久化记忆的版本。

需要提前说明的是：下文涉及的各家免费额度、模型名称、SDK 版本都属于容易变动的信息，请以各服务官网当前公布的条件为准。

准备工作

动手之前先把账号和凭据备齐，后面每一步都会用到。

Google Cloud 账号：需要一个绑定了结算账号的项目。即使全程只用免费额度，项目本身仍然要关联结算账号才能开通相关服务。
LINE Developers 账号：有 LINE 账号即可注册，用来创建 Messaging API 渠道。
Google AI Studio 账号：用 Google 账号登录后签发一个 Gemini API Key。
本地环境：安装好 gcloud CLI 并完成登录，能执行 gcloud 命令；Python 版本与部署时选择的运行时保持一致。

整体架构分成四层：LINE 平台负责把用户消息以 HTTPS POST 的形式推给你的函数；Cloud Run functions 承担 Webhook 接收与流程控制；Cloud Firestore 存放每个用户的对话历史；Gemini API 负责生成回复。函数本身是无状态的，上下文完全靠 Firestore 读写来维持，这一点是整套设计的关键。

操作步骤

第一步：启用所需的 Google Cloud API

在 Cloud Shell 或本地终端执行下面这条命令，一次性打开后续会用到的接口：

gcloud services enable \
  artifactregistry.googleapis.com \
  cloudbuild.googleapis.com \
  cloudfunctions.googleapis.com \
  run.googleapis.com \
  firestore.googleapis.com

其中 Artifact Registry 与 Cloud Build 是第 2 代函数构建镜像时的依赖，Run 是底层承载服务，Firestore 用于存储对话历史。

第二步：创建 Firestore 数据库

在 Google Cloud 控制台搜索并进入 Firestore，如果项目里还没有数据库，点击创建数据库：

数据库类型选择原生模式。
数据库 ID 保持默认的 (default) 不变。
位置选择离你较近的区域，例如 asia-northeast1。
点击创建。

数据库位置一旦确定通常不便更改，建议和后面部署函数时使用的区域保持一致，减少跨区访问带来的延迟。

第三步：准备 LINE 官方账号与 Messaging API 渠道

Channel Secret：位于「渠道基本设置」标签页下方，用于校验 Webhook 请求签名。
Channel Access Token（长期）：位于「Messaging API 设置」标签页最下方，点击签发按钮获取，用于调用回复接口。

这两个值后面会作为环境变量注入函数，不要写死在代码里。

第四步：编写函数代码

新建一个工作目录，放入两个文件。先是依赖清单 requirements.txt：

functions-framework==3.*
line-bot-sdk==3.*
google-genai>=1.0.0
google-cloud-firestore==2.*

这里用的是 Google 的统一 SDK google-genai。早期常用的 google-generativeai 已经不再维护，新项目应当直接采用新 SDK 的写法。

接着是主程序 main.py，可以分成初始化、Webhook 入口、历史读取、历史保存、消息处理五块来看。

初始化与配置：从环境变量读取三个凭据，构造 LINE 的配置对象与 Webhook 处理器，创建 Gemini 客户端和 Firestore 客户端。在 Cloud Run functions 环境中，Firestore 客户端会自动解析身份，无需手动传入密钥文件。同时定义系统提示词和保留的历史轮数上限。

import os
import time
import functions_framework
from google.cloud import firestore
from google import genai
from google.genai import types
from linebot.v3.exceptions import InvalidSignatureError
from linebot.v3.messaging import (
    ApiClient, Configuration, MessagingApi,
    ReplyMessageRequest, TextMessage,
)
from linebot.v3.webhook import WebhookHandler
from linebot.v3.webhooks import MessageEvent, TextMessageContent

CHANNEL_ACCESS_TOKEN = os.environ.get('LINE_CHANNEL_ACCESS_TOKEN')
CHANNEL_SECRET = os.environ.get('LINE_CHANNEL_SECRET')
GEMINI_API_KEY = os.environ.get('GEMINI_API_KEY')

configuration = Configuration(access_token=CHANNEL_ACCESS_TOKEN)
handler = WebhookHandler(CHANNEL_SECRET)
client = genai.Client(api_key=GEMINI_API_KEY)
MODEL_NAME = 'gemini-3.5-flash'

SYSTEM_PROMPT = """你是帮助用户的贴心 AI 助手。
请遵守以下规则：
1. 回复长度适合在聊天窗口阅读，适当换行。
2. 结合此前的对话上下文作答。
3. 用通俗的方式解释专业术语。
"""

db = firestore.Client()
HISTORY_LIMIT = 10

Webhook 入口：函数以 HTTP 触发，从请求头取出签名，把请求体交给处理器校验。签名不匹配返回 400，其他异常返回 500，正常情况返回 200。

@functions_framework.http
def line_webhook(request):
    signature = request.headers.get('X-Line-Signature')
    body = request.get_data(as_text=True)
    try:
        handler.handle(body, signature)
    except InvalidSignatureError:
        return 'Invalid signature', 400
    except Exception as e:
        print(f"Internal Error: {e}")
        return 'Internal Server Error', 500
    return 'OK', 200

读取历史：以用户 ID 作为文档 ID，从 users 集合中取出保存的对话数组，再转换成 Gemini SDK 要求的消息结构。用户发言对应 user 角色，模型回复对应 model 角色，按时间顺序依次追加。

def get_chat_history(user_id: str):
    doc_ref = db.collection('users').document(user_id)
    doc = doc_ref.get()
    history = []
    if doc.exists:
        saved_history = doc.to_dict().get('history', [])
        for turn in saved_history:
            history.append(types.Content(
                role="user", parts=[types.Part(text=turn['user'])]))
            history.append(types.Content(
                role="model", parts=[types.Part(text=turn['model'])]))
    return history

保存历史：把本轮问答追加进数组，附带时间戳；超过上限时只保留最近的若干轮，再写回文档。使用 merge=True 可以避免覆盖文档中的其他字段。

def save_chat_history(user_id: str, user_text: str, model_text: str):
    doc_ref = db.collection('users').document(user_id)
    doc = doc_ref.get()
    current_history = []
    if doc.exists:
        current_history = doc.to_dict().get('history', [])
    current_history.append({
        "user": user_text,
        "model": model_text,
        "timestamp": time.time(),
    })
    if len(current_history) > HISTORY_LIMIT:
        current_history = current_history[-HISTORY_LIMIT:]
    doc_ref.set({'history': current_history}, merge=True)

消息处理：注册到处理器上的回调负责串起整条流程——取用户 ID 与消息文本、读历史、创建带系统提示词的聊天会话、发送消息拿到回复、写回历史、最后通过 LINE 的回复接口把文本发出去。

@handler.add(MessageEvent, message=TextMessageContent)
def handle_message(event):
    user_id = event.source.user_id
    user_message = event.message.text
    try:
        history = get_chat_history(user_id)
        chat = client.chats.create(
            model=MODEL_NAME,
            history=history,
            config=types.GenerateContentConfig(
                system_instruction=SYSTEM_PROMPT),
        )
        response = chat.send_message(user_message)
        reply_text = response.text
        save_chat_history(user_id, user_message, reply_text)
        with ApiClient(configuration) as api_client:
            line_bot_api = MessagingApi(api_client)
            line_bot_api.reply_message_with_http_info(
                ReplyMessageRequest(
                    reply_token=event.reply_token,
                    messages=[TextMessage(text=reply_text)],
                )
            )
    except Exception as e:
        print(f"Error handling message: {e}")

第五步：部署到 Cloud Run functions

在代码目录下执行部署命令，指定第 2 代、Python 3.11 运行时、区域、入口函数名，并把三个凭据作为环境变量传入：

gcloud functions deploy line-gemini-bot \
  --gen2 \
  --runtime=python311 \
  --region=asia-northeast1 \
  --source=. \
  --entry-point=line_webhook \
  --trigger-http \
  --allow-unauthenticated \
  --set-env-vars LINE_CHANNEL_ACCESS_TOKEN="<渠道访问令牌>",LINE_CHANNEL_SECRET="<渠道密钥>",GEMINI_API_KEY="<Gemini API Key>"

部署完成后，命令输出里会给出函数的访问地址，下一步要用到。

第六步：配置 LINE 侧的 Webhook 与自动应答

回到 LINE Developers 控制台的「Messaging API 设置」标签页：

点击 Webhook URL 的编辑按钮，填入上一步得到的函数地址并保存。
点击验证按钮，确认显示成功。
把「使用 Webhook」开关打开。

还有一步容易被忽略：进入 LINE Official Account Manager，打开右上角设置中的「应答设置」，把详细设置里的「应答消息」关闭。如果不关，机器人的回复会和 LINE 的默认自动应答同时发出，用户会收到两条消息。

一个完整示例

把上面的步骤串起来跑一遍，验证上下文是否真的生效。

在「Messaging API 设置」标签页找到二维码，用手机扫码把机器人加为好友，然后发送两条消息：

用户：「我最喜欢的云服务是 Google Cloud」
机器人：「Google Cloud 呀！你最喜欢它的哪个服务呢？」
用户：「我刚才说我喜欢的云服务是什么来着？」
机器人：「是 Google Cloud，你刚才告诉过我。」

如果第二轮追问能得到基于前文回答的结果，说明 Firestore 的读写、历史拼接和 Gemini 调用这条链路已经打通。此时可以打开 Firestore 控制台，在 users 集合下找到以用户 ID 命名的文档，里面应当能看到 history 数组，每一项包含 user、model 和 timestamp 三个字段。

想调整助手性格，改 SYSTEM_PROMPT 即可；想让它记住更多轮对话，调大 HISTORY_LIMIT，但要注意历史越长，每次请求消耗的输入 token 越多，响应也会变慢。

注意事项

执行身份需要 Firestore 权限。第 2 代函数默认使用一个计算服务账号运行，通常是「项目编号-compute@developer.gserviceaccount.com」。这个账号必须拥有 Firestore 的读写权限。如果调用时报权限错误，到控制台的「IAM 与管理」中确认该服务账号已被授予「Cloud Datastore 用户」角色。

环境变量会留在命令历史里。上面的部署命令为了演示方便，把凭据直接写在 --set-env-vars 中，值会残留在 shell 历史记录里。正式使用或需要与他人共享的环境，应当先把凭据存入 Secret Manager，再通过 --set-secrets 选项引用。

并发写入可能丢历史。保存历史的逻辑是「读取 → 追加 → 写入」三步，如果同一个用户在极短时间内连续发送多条消息，两次调用可能互相覆盖，导致部分记录丢失。这是面向个人自用的简化实现，若预期会有并发请求，需要改用事务处理（例如 Firestore 的事务装饰器）来保证一致性。

免费额度有上限且会调整。函数调用次数、Firestore 的每日读写次数与存储量、Gemini 免费层的速率限制、LINE 消息推送条数，都有各自的额度约束，超出后可能产生费用或被限流。具体数值请以各服务官网当前公布的信息为准，不要以本文描述为准。

模型名称与 SDK 版本会变。代码中的模型标识和依赖版本号需要按实际可用情况调整，升级 SDK 时留意接口签名是否有变化。

这套基座还能扩展成主动推送。除了被动应答，同样的函数与数据存储可以配合定时任务，在固定时间主动向用户推送内容，例如每天早晨发送一道练习题或一段资讯摘要。这类扩展需要额外的调度组件，属于本教程范围之外的下一步。