File size: 2,432 Bytes
55b1be9 34287b9 f23a640 55b1be9 2cf7795 76eaf5a 4db0db1 846eaf3 f23a640 55b1be9 df5bf2a 55b1be9 7a8d7b0 9a45992 0385c62 2cf7795 55b1be9 6482098 55b1be9 2cf7795 6482098 34287b9 0a52d39 3a2e91a 34287b9 3a2e91a 34287b9 0385c62 2cf7795 4a7f8d2 3a2e91a 4a7f8d2 2cf7795 4a7f8d2 3a2e91a f23a640 4a7f8d2 f23a640 4a7f8d2 619c607 3bf6983 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 |
from fastapi.middleware.cors import CORSMiddleware
from fastapi import FastAPI, Request, Header, BackgroundTasks, HTTPException, status
import google.generativeai as genai
import json
import os
from linebot import (
LineBotApi, WebhookHandler
)
from linebot.exceptions import (
InvalidSignatureError
)
from linebot.models import (
MessageEvent, TextMessage, TextSendMessage, ImageSendMessage, AudioMessage
)
GOOGLE_API_KEY = os.environ["GOOGLE_API_KEY"]
genai.configure(api_key = GOOGLE_API_KEY)
# Initialize the Gemini model
model = genai.GenerativeModel('gemini-pro')
app = FastAPI()
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
line_bot_api = LineBotApi(os.environ["CHANNEL_ACCESS_TOKEN"])
line_handler = WebhookHandler(os.environ["CHANNEL_SECRET"])
working_status = os.getenv("DEFALUT_TALKING", default = "true").lower() == "true"
@app.get("/")
def root():
return {"title": "Echo Bot"}
@app.post("/webhook")
async def webhook(
request: Request,
background_tasks: BackgroundTasks,
x_line_signature=Header(None),
):
body = await request.body()
try:
background_tasks.add_task(
line_handler.handle, body.decode("utf-8"), x_line_signature
)
except InvalidSignatureError:
raise HTTPException(status_code=400, detail="Invalid signature")
return "ok"
@line_handler.add(MessageEvent, message=TextMessage)
def handle_message(event):
global working_status
if event.message.text == "再見":
working_status = True
line_bot_api.reply_message(
event.reply_token,
TextSendMessage(text="Bye!"))
return
messages = []
if working_status:
message = event.message.text
messages.append({
"role": "user",
"parts": [message],
})
response = model.generate_content(messages)
out = response.parts[0].text
# Check if response is valid
if not response.parts or not response.parts[0].text:
out = "Gemini: [No response or the response does not contain text]"
line_bot_api.reply_message(
event.reply_token,
TextSendMessage(text=out))
if __name__ == "__main__":
uvicorn.run("main:app", host="0.0.0.0", port=7860, reload=True) |