今天撰寫了串聯Line Bot和Gemini的程式,在撰寫前我參考了網路上其他人撰寫的程式,有些是利用ngrok串聯,有些是串聯到Open AI,每個人最後製作出的程式都有很大的差別,但都有著一定的相似性。
以下是找到的一些程式碼還有其中一些相似的地方
from flask import Flask, request, abort from linebot import ( LineBotApi, WebhookHandler ) from linebot.exceptions import ( InvalidSignatureError ) from linebot.models import * #======python的函數庫========== import tempfile, os import datetime import openai import time import traceback #======python的函數庫========== app = Flask(__name__) static_tmp_path = os.path.join(os.path.dirname(__file__), 'static', 'tmp') # Channel Access Token line_bot_api = LineBotApi(os.getenv('CHANNEL_ACCESS_TOKEN')) # Channel Secret handler = WebhookHandler(os.getenv('CHANNEL_SECRET')) # OPENAI API Key初始化設定 openai.api_key = os.getenv('OPENAI_API_KEY') def GPT_response(text): # 接收回應 response = openai.Completion.create(model="gpt-3.5-turbo-instruct", prompt=text, temperature=0.5, max_tokens=500) print(response) # 重組回應 answer = response['choices'][0]['text'].replace('。','') return answer # 監聽所有來自 /callback 的 Post Request @app.route("/callback", methods=['POST']) def callback(): # get X-Line-Signature header value signature = request.headers['X-Line-Signature'] # get request body as text body = request.get_data(as_text=True) app.logger.info("Request body: " + body) # handle webhook body try: handler.handle(body, signature) except InvalidSignatureError: abort(400) return 'OK' # 處理訊息 @handler.add(MessageEvent, message=TextMessage) def handle_message(event): msg = event.message.text try: GPT_answer = GPT_response(msg) print(GPT_answer) line_bot_api.reply_message(event.reply_token, TextSendMessage(GPT_answer)) except: print(traceback.format_exc()) line_bot_api.reply_message(event.reply_token, TextSendMessage('你所使用的OPENAI API key額度可能已經超過,請於後台Log內確認錯誤訊息')) @handler.add(PostbackEvent) def handle_message(event): print(event.postback.data) @handler.add(MemberJoinedEvent) def welcome(event): uid = event.joined.members[0].user_id gid = event.source.group_id profile = line_bot_api.get_group_member_profile(gid, uid) name = profile.display_name message = TextSendMessage(text=f'{name}歡迎加入') line_bot_api.reply_message(event.reply_token, message) import os if __name__ == "__main__": port = int(os.environ.get('PORT', 5000)) app.run(host='0.0.0.0', port=port)
from flask import Flask, request, abort
from linebot import WebhookHandler, LineBotApi
from linebot.exceptions import InvalidSignatureError
from linebot.models import MessageEvent, TextMessage, TextSendMessage
import requests
import openai
from pyngrok import ngrok
from pyngrok.conf import PyngrokConfig
# 初始化 Flask 和 LineBot 相关实例
app = Flask(__name__)
line_bot_api = LineBotApi(LINE_ACCESS_TOKEN)
handler = WebhookHandler(LINE_CHANNEL_SECRET)try:
# ngrok 连接
webhook_url = ngrok.connect(addr="127.0.0.1:5000", pyngrok_config=PyngrokConfig(start_new_session=True))
print("Ngrok Tunnel URL:", webhook_url)
except Exception as e:
print("Error while connecting with ngrok:", e)# LINE Bot 接收消息的路由
@app.route("/", methods=["POST"])
def callback():
signature = request.headers["X-Line-Signature"]
body = request.get_data(as_text=True)
try:
handler.handle(body, signature)
except InvalidSignatureError:
abort(400)
return "OK"# LINE Bot 处理文本消息的逻辑
@handler.add(MessageEvent, message=TextMessage)
def handle_message(event):
message = event.message.text # 处理天气查询功能
messages = message.split()
country = ["蘇澳鎮", "頭城鎮", "宜蘭市", "南澳鄉", "羅東鎮", "三星鄉", "大同鄉", "五結鄉", "員山鄉", "冬山鄉", "礁溪鄉", "壯圍鄉"]
if messages[0] in country:
base_url = "https://opendata.cwa.gov.tw/api/v1/rest/datastore/F-D0047-003?Authorization=CWA-0C63261F-E152-485F-B43E-F17EE1CADF6B"
response = requests.get(base_url)
if response.status_code == 200:
data = response.json()
for location in data["records"]["locations"][0]["location"]:
if location["locationName"] == messages[0]:
for weather_element in location["weatherElement"]:
for time in weather_element["time"]:
if time["startTime"].startswith("2024-04-25"):
weather_info = f"{weather_element['description']}:{time['elementValue'][0]['value']}"
line_bot_api.reply_message(event.reply_token, TextSendMessage(text=weather_info))
return
line_bot_api.reply_message(event.reply_token, TextSendMessage(text="未找到相应的天气信息"))
else:
line_bot_api.reply_message(event.reply_token, TextSendMessage(text=f"Error: {response.status_code}"))
else:
line_bot_api.reply_message(event.reply_token, TextSendMessage(text="您输入的数据有误!"))
# 主程序入口
if __name__ == "__main__":
app.run()
# 安裝 Google AI Python SDK
# $ pip install google-generativeai
# 參考資料:https://ai.google.dev/gemini-api/docs/get-started/python
import os
import google.generativeai as genai
# 設定 API Key
genai.configure(api_key=os.environ["GEMINI_API_KEY"])
# 設定生成配置
generation_config = {
"temperature": 1,
"top_p": 0.95,
"top_k": 64,
"max_output_tokens": 8192,
"response_mime_type": "text/plain",
}
# 設定安全設置
safety_settings = [
{
"category": "HARM_CATEGORY_HARASSMENT",
"threshold": "BLOCK_LOW_AND_ABOVE",
},
{
"category": "HARM_CATEGORY_HATE_SPEECH",
"threshold": "BLOCK_LOW_AND_ABOVE",
},
{
"category": "HARM_CATEGORY_SEXUALLY_EXPLICIT",
"threshold": "BLOCK_LOW_AND_ABOVE",
},
{
"category": "HARM_CATEGORY_DANGEROUS_CONTENT",
"threshold": "BLOCK_LOW_AND_ABOVE",
},
]
# 創建模型
model = genai.GenerativeModel(
model_name="gemini-1.5-flash",
safety_settings=safety_settings,
generation_config=generation_config,
在這次撰寫程式中引入了許多之前沒用過的函數,如果有任何錯誤歡迎大家指教,謝謝閱讀這篇文章,明天會進行測試和除錯,希望程式沒有大問題。