23.12.06.(WED).TIL.
ChatGPT API 활용 카카오톡 챗봇
ChatGPT API 활용 카카오톡 챗봇
##### 기본 정보 설정 단계 #####
from fastapi import Request, FastAPI
from openai import OpenAI
import threading
import time
import queue as q
import os
# OpenAI API 키
client = OpenAI(
api_key=os.environ['YOUR_API_KEY'], # this is also the default, it can be omitted
)
##### 기능 함수 구현 단계 #####
# 메시지 전송
def textResponseFormat(bot_response):
response = {"version":"2.0", "template":{"outputs":[{"simpleText":{"text":bot_response}}], "quickReplies":[]}}
return response
# 사진 전송
def imageResponseFormat(bot_response, prompt):
output_text = prompt + "내용에 관한 이미지입니다"
response = {"version":"2.0", "template":{"outputs":[{"simpleImage":{"imageUrl":bot_response, "altText":output_text}}], "quickReplies":[]}}
return response
# 응답 초과 시 답변
def timeover():
response = { "version":"2.0", "template":{
"outputs":[
{
"simpleText":{
"text":"아직 제가 생각이 끝나지 않았어요 🙏🙏 \n잠시 후 아래 말풍선을 눌러주세요 👆"
}
}
],
"quickReplies":[
{
"action":"message",
"label":"생각 다 끝났나요? 🙋♂️",
"messageText":"생각 다 끝났나요?"
}
]
}}
return response
# ChatGPT에게 질문/답변받기
def getTextFromGPT(prompt):
messages_prompt = [{"role": "system", "content": "You are a thoughtful assistant. Respond to all input in 25 words and answer in Korea"}]
messages_prompt += [{"role": "system", "content": prompt}]
response = client.chat.completions.create(model="gpt-3.5-turbo", messages=messages_prompt)
message = response["choices"][0]["message"]["content"]
return message
# DALL.E 2에게 질문/그림 URL 받기
def getImageURLFromDALLE(prompt):
response = openai.Image.create(prompt=prompt, n=1, size="512x512")
image_url = response["data"][0]["url"]
return image_url
# 텍스트 파일 초기화
def dbReset(filename):
with open(filename, 'w') as f:
f.write("")
###### 서버 생성 단계 #####
app = FastAPI()
@app.get("/")
async def root():
return {"message": "kakaoTest"}
@app.post("/chat/")
async def chat(request: Request):
kakaorequest = await request.json()
return mainChat(kakaorequest)
##### 메인 함수 구현 단계 ######
# 메인 함수
def mainChat(kakaorequest):
run_flag = False
start_time = time.time()
# 응답 결과를 저장하기 위한 텍스트 파일 생성
cwd = os.getcwd()
filename = cwd + "/botlog.txt"
if not os.path.exists(filename):
with open(filename, "w") as f:
f.write("")
else:
print("File Exists")
# 답변 생성 함수 실행
response_queue = q.Queue()
request_respond = threading.Thread(target=responseOpenAI, args = (kakaorequest, response_queue, filename))
request_respond.start()
# 답변 생성 시간 체크
while (time.time() - start_time < 3.5):
if not response_queue.empty():
# 3.5초 안에 답변이 완성되면 바로 값을 반환
response = response_queue.get()
run_flag = True
break
# 안정적인 구동을 위한 딜레이 타임 설정
time.sleep(0.01)
# 3.5초 안에 답변이 생성되지 않을 경우
if run_flag == False:
response = timeover()
return response
# 답변/사진 요청 및 응답 확인 함수
def responseOpenAI(request, response_queue, filename):
# 사용자가 버튼을 클릭하여 답변 완성 여부를 다시 봤을 시
if '생각 다 끝났나요?' in request["userRequest"]["utterance"]:
# 텍스트 파일 열기
with open(filename) as f:
last_update = f.read()
# 텍스트 파일 내 저장된 정보가 있을 경우
if len(last_update.split()) > 1:
kind, bot_res, prompt = last_update.split()[0], last_update.split()[1], last_update.split()[2]
if kind == "img":
response_queue.put(imageResponseFormat(bot_res, prompt))
else:
response_queue.put(textResponseFormat(bot_res))
dbReset(filename)
# 이미지 생성을 요청한 경우
elif '/img' in request["userRequest"]["utterance"]:
dbReset(filename)
prompt = request["userRequest"]["utterance"].replace("/img", "")
bot_res = getImageURLFromDALLE(prompt)
response_queue.put(imageResponseFormat(bot_res, prompt))
save_log = "img" + " " + str(bot_res) + " " + str(prompt)
with open(filename, 'w') as f:
f.write(save_log)
# ChatGPT 답변을 요청한 경우
elif '/ask' in request["userRequest"]["utterance"]:
dbReset(filename)
prompt = request["userRequest"]["utterance"].replace("/ask", "")
bot_res = getTextFromGPT(prompt)
response_queue.put(textResponseFormat(bot_res))
save_log = "ask" + " " + str(bot_res) + " " + str(prompt)
with open(filename, 'w') as f:
f.write(save_log)
# 아무 답변 요청이 없는 채팅일 경우
else:
# 기본 response 값
base_response = {'version':'2.0', 'template': {'outputs': [], 'quickReplies': []}}
response_queue.put(base_response)
Error #01
출력 : {"detail":"Not Found"}
오류 메시지 : GET http://127.0.0.1:8000/chat/ 404 chat/:1 (Not Found)
원인 : app = FastAPI()가 두 번 정의되어 있었기 때문
Error #02
출력 : {"detail":"Method Not Allowed"}
오류 메시지 : GET http://127.0.0.1:8000/chat/ 405 chat/:1 (Method Not Allowed)
해결 : 코드 수정
from fastapi import Request, FastAPI
app = FastAPI()
# `/` 경로에 대한 처리기 등록
@app.get("/")
async def root():
return {"message": "kakaoTest"}
@app.post("/chat/")
async def chat(request: Request):
kakaorequest = await request.json()
print(kakaorequest)
return {"message": "kakaoTest"}
Error #03
출력 : {Internal Server ERror}
오류 메시지 : GET http://127.0.0.1:8000/chat/ 500 chat/:1 (Internal Server Error)
원인 :
코드를 살펴보니 `@app.rout("/chat/", methods=["GET", "POST"])` 데코레이터와 함께 함수를 정의하고 있습니다. 그러나 FastAPI에서는 `@app.route` 대신에 `@app.get` 또는 `app.post`를 사용하여 각각의 HTTP 메서드에 대한 핸들러를 등록하는 것이 권장됩니다.
또한, `@app.route`를 사용할 경우에는 `methods` 매개변수 대신에 각 HTTP 메서드에 대한 데코레이터를 별도로 사용해야 합니다.
아래는 수정된 코드입니다.
from fastapi import Request, FastAPI
app = FastAPI()
# `/` 경로에 대한 처리기 등록
@app.get("/")
async def root():
return {"message": "kakaoTest"}
# `/chat/` 경로에 대한 GET 요청 처리기 등록
@app.get("/chat/")
async def get_chat():
return {"message": "kakaoTest"}
# `/chat/` 경로에 대한 POST 요청 처리기 등록
@app.post("/chat/")
async def post_chat(request: Request):
kakaorequest = await request.json()
print(kakaorequest)
return {"message": "kakaoTest"}
