在前兩篇測試中已在 render.com 上使用 serverless 函式執行平台完成 LINE Bot 串接 LLM 模型的初步測試, 雖然可用但程式仍很粗糙, 本篇旨在繼續精進此聊天機器人程式. 本系列之前的測試文章參考 :
最近在測試用 LINE Bot 串接 LLM (GPT 或 Gemini) 時發現之前的同步處理有阻塞問題, 有時提問後卻沒有得到 AI 回覆, 這是因為呼叫 LLM API 生成回覆通常需要數秒至十數秒, 這段期間主執行緒會被阻塞, 而 LINE 平台對 webhook 的 HTTP 回應有時間限制 (通常約為 1 秒), 若伺服器未能在時限內回傳 HTTP 200 狀態碼, LINE 會視為 webhook 呼叫失敗, 導致使用者提問後收不到任何回覆.
解決辦法是利用執行緒改為非同步處理, 作法是 Flask webhook 一收到訊息先立即回傳 200 OK (非同步處理, 非阻塞等待), 再啟動一個背景執行緒呼叫 LLM, 等收到 LLM 生成之結果再透過 push_message() 將結果回傳給 LINE 平台回覆使用者, 程序如下圖所示 :
Python 內建 threading 套件來實作多執行緒功能, 使用前需先匯入 threading :
import threading
然後定義要在背景中執行的函式 (例如此處是要串接 LLM) :
def background_task(args):
do_something
接下來呼叫 threading.Thread() 並傳入要執行之函式與所需參數 (tuple) 來建立執行緒物件 :
thread.Thread(background_task, args=(arg1, arg2, ...))
最後呼叫 Thread 物件的 start() 來啟動執行緒 :
thread.start()
以下是將前一篇串接 Gemini 的程式利用 threading 改為非同步的版本 :
# linebot_gemini.py
from flask import abort
from linebot import LineBotApi, WebhookHandler
from linebot.exceptions import InvalidSignatureError
from linebot.models import MessageEvent, TextMessage, TextSendMessage
import os
import google.generativeai as genai
import threading
def main(request, **kwargs):
# 取得主程式傳入的 LINE 與 Gemini 金鑰
config=kwargs.get('config', {})
secret=config.get('LINE_CHANNEL_SECRET')
token=config.get('LINE_CHANNEL_ACCESS_TOKEN')
gemini_api_key=config.get('GEMINI_API_KEY', os.getenv('GEMINI_API_KEY'))
# 檢查必要參數
if not secret or not token or not gemini_api_key:
return {'error': 'Missing LINE or Gemini credentials'}
# 初始化 LINE API 與 Webhook Handler
line_bot_api=LineBotApi(token)
handler=WebhookHandler(secret)
# 設定 Gemini API 金鑰
genai.configure(api_key=gemini_api_key)
# 建立一個在背景執行 Gemini 並推送訊息的函式
def ask_gemini(user_id, user_text):
system_instruction='你是一個繁體中文AI助理, 請以台灣人的習慣用語回答'
try:
# 每次呼叫時都建立模型物件以確保執行緒安全
model=genai.GenerativeModel(
'gemini-2.5-flash',
system_instruction=system_instruction
)
# 呼叫 Gemini API (會在背景執行)
response=model.generate_content(user_text)
reply_text=response.text.strip() if response.text else '(無法取得回覆)'
except Exception as e:
reply_text = f'處理您的請求時發生錯誤：{e}'
# 使用 Push API 將 AI 生成結果回覆給使用者
line_bot_api.push_message(
user_id,
TextSendMessage(text=reply_text)
)
# 註冊 LINE 訊息事件 (偵測詢問)
@handler.add(MessageEvent, message=TextMessage)
def handle_message(event):
user_text=event.message.text # 取得使用者之詢問訊息
user_id=event.source.user_id # 取得使用者的 user_id
# 先回覆一個處理中的訊息避免逾時
line_bot_api.reply_message(
event.reply_token,
TextSendMessage(text='好的，請稍候 ...')
)
# 建立一個新的執行緒來處理可能耗時的生成任務
thread=threading.Thread(target=ask_gemini, args=(user_id, user_text))
thread.start()
# 驗證簽章
signature=request.headers.get('X-Line-Signature', '')
body=request.get_data(as_text=True)
try:
handler.handle(body, signature)
except InvalidSignatureError:
abort(400, 'Invalid signature')
# main() 立即傳回 'ok' 不等待 Gemini 回應
return {'status': 'ok'}
此程式關鍵之處在於當偵測到使用者的 LINE 詢問時會啟動一個執行緒在背景執行串接 Gemini 的任務, 主執行緒則立即先傳回 {'status': 'ok'} (即 200 OK) 給 LINE Messaging 伺服器以免逾時. 等收到 Gemini 生成之回覆時, 背景執行緒會呼叫 LINE Bot API 的 push_message() 方法傳回結果給 LINE 平台回覆使用者.
其次, 此處在建立 GenerativeModel 物件時傳入 system_instruction 參數來設定 System role, 這樣就可以避免 Gemini 用殘體中文來回應了.
注意, 此程式中使用了 LINE Bot API 的兩個函式 reply_message() 與 push_message(), 這兩個函式用途不同, 不要誤用了:
- reply_message() :
當 LINE 使用者傳送訊息時, LINE 伺服器會觸發 webhook, 並在 event 裡面附帶一個 replyToken, 這個 token 只能使用一次且有效時間只有幾秒 (小於 1 分鐘), 程式需要在 webhook 的執行流程內用這個 token 呼叫 reply_message() 來傳回使用者期望立即得到的回應. - push_message() :
伺服器可在任何時間呼叫 push_message() 主動發訊息給使用者或群組, 只要知道對方的 user_id, group_id 或 room_id 即可無次數限制隨時推送訊息 (不需 replyToken, 也不受 webhook timeout 影響).
兩者特性比較如下表所示 :
| 特性 | reply_message() | push_message() |
|---|---|---|
| 觸發來源 | 用戶傳訊息後，LINE 平台發送 Webhook 時提供的 reply_token | 伺服器主動發送，不需使用者觸發 |
| 使用場景 | 回覆使用者訊息（互動對話） | 主動通知、推播訊息、定時發送 |
| 是否需要 reply_token | 需要（由 LINE Webhook 提供，有效期約 1 分鐘） | 不需要 |
| 是否可主動發送 | 否，只能在事件回應時使用 | 可，由伺服器主動呼叫 API 發送 |
| 是否需等待使用者互動 | 是 | 否 |
| 一次可傳送訊息數量 | 最多 5 則訊息 | 最多 5 則訊息 |
| 速率限制 | 受 reply_token 有效時間限制（約 1 分鐘） | 依帳號方案而異： 免費帳號每日 500 則 付費帳號依方案增加 |
| 典型應用範例 | 使用者問「天氣如何？」→ Bot 回覆天氣資訊 | 每天早上 8 點主動推播天氣預報 |
| 回應時限 | 需於 webhook 收到後 1 分鐘內回覆 | 無時限，可隨時發送 |
| 是否可在 Verify 測試中使用 | 無效（verify 測試不提供 reply_token） | 可獨立測試推播功能 |
| 需要的權限範圍 | Messaging API 基本權限 | 需開啟「Push message」權限（Messaging API plan） |
將此 LINE Bot 程式貼到 render.com 更新 linebot_gemini.py 模組 :
取得 webhook 網址後, 登入 LINE Business 網站, 更新 LINE Messaging 的 Webhook 網址 :
結果如下 :
當偵測到使用者提問時會先呼叫 LINE Bot API 的 reply_message() 方法傳回 "好的，請稍候 ..." 訊息是可有可無的, 主要是讓使用者知道提問正在處理中, 避免使用者以為已讀不回.
下面是 GPT 串接程式的非同步版 :
# linebot_gpt.py
from flask import abort
from linebot import LineBotApi, WebhookHandler
from linebot.exceptions import InvalidSignatureError
from linebot.models import MessageEvent, TextMessage, TextSendMessage
import os
from openai import OpenAI
import threading
def main(request, **kwargs):
# 取得主程式傳入的 LINE 與 OpenAI 金鑰權杖
config=kwargs.get('config', {})
secret=config.get('LINE_CHANNEL_SECRET')
token=config.get('LINE_CHANNEL_ACCESS_TOKEN')
openai_api_key=config.get('OPENAI_API_KEY', os.getenv('OPENAI_API_KEY'))
# 檢查必要參數
if not secret or not token or not openai_api_key:
return {'error': 'Missing LINE or OpenAI credentials'}
# 初始化 LINE API 與 Webhook Handler
line_bot_api=LineBotApi(token)
handler=WebhookHandler(secret)
# 設定 OpenAI API 金鑰
client=OpenAI(api_key=openai_api_key)
# 建立一個在背景執行串接 GPT 並推送訊息的函式
def ask_gpt(user_id, user_text):
system_instruction='你是一個繁體中文AI助理, 請以台灣人的習慣用語回答'
try:
# 呼叫 OpenAI GPT 生成回應
response=client.chat.completions.create(
model='gpt-3.5-turbo', # GPT 模型
messages=[
{'role': 'system', 'content': '你是一個繁體中文AI助理'},
{'role': 'user', 'content': user_text}
],
#max_tokens=300
)
reply_text=response.choices[0].message.content
except Exception as e:
reply_text = f'處理您的請求時發生錯誤：{e}'
# 使用 Push API 將 AI 生成結果回覆給使用者
line_bot_api.push_message(
user_id,
TextSendMessage(text=reply_text)
)
# 註冊 LINE 訊息事件 (偵測詢問)
@handler.add(MessageEvent, message=TextMessage)
def handle_message(event):
user_text=event.message.text # 取得使用者之詢問訊息
user_id=event.source.user_id # 取得使用者的 user_id
# 先回覆一個處理中的訊息避免逾時
line_bot_api.reply_message(
event.reply_token,
TextSendMessage(text='好的，請稍候 ...')
)
# 建立一個新的執行緒來處理可能耗時的生成任務
thread=threading.Thread(target=ask_gpt, args=(user_id, user_text))
thread.start()
# 驗證簽章
signature=request.headers.get('X-Line-Signature', '')
body=request.get_data(as_text=True)
try:
handler.handle(body, signature)
except InvalidSignatureError:
abort(400, 'Invalid signature')
# main() 立即傳回 'ok' 不等待 Gemini 回應
return {'status': 'ok'}
結構一樣, 只是改成 OpenAI 而已 (System role 的帶入方式也不同), 測試結果如下 :
哈哈, gpt-turbo-3.5 真的太老了, 沒有用 function calling 去 callout 真的會亂講.
已更新 GitHub 上的 serverless 儲存庫 (repo), 將上面兩個 LINE Bot webhook 程式直接放進 functions 資料夾內, 這樣萬一 render.com 上的 serverless 平台重啟時, linebot_gpt.py 與 linebot_gemini.py 就已存在, 毋須再手動新增.
沒有留言 :
張貼留言