🪴 Quartz 4.0

Search

SearchSearch

2. Flask 애플리케이션 개발

Aug 09, 20249 min read

3.1 기본 애플리케이션 구조 설정

구름 ide에서 아래와 같이 파일을 만듭니다.

부산사투리챗봇/
├── app.py
├── requirements.txt
└── .env

각 파일의 역할:

  • app.py: 메인 애플리케이션 코드
  • requirements.txt: 필요한 파이썬 패키지 목록
  • .env: 환경 변수 저장

3.2 필요한 라이브러리 설치

터미널에서 다음 명령어를 실행하여 필요한 라이브러리를 설치합니다:

pip install flask openai python-dotenv requests flask-cors flask-limiter

requirements.txt 파일에 다음 내용을 추가합니다:

  1. 우측 마우스 버튼을 클릭하여 팝업에서 새 파일을 선택합니다.
  2. 만들어진 새파일에서 requirements.txt 파일명을 입력합니다.
  3. 파일 내용에 아래 내용을 붙여넣으세요.
flask==2.0.1
openai==0.27.0
python-dotenv==0.19.0
requests==2.26.0
flask-cors==3.0.10
flask-limiter==2.4.0

3.3 기본 Flask 애플리케이션 코드 작성

위 내용처럼 app.py파일을 만들고, app.py 파일에 다음 코드를 복사 붙여넣기 합니다. (기본적으로 있는 py파일을 이름 변경없이 그대로 사용하셔도 됩니다.)

from flask import Flask, request, jsonify
import logging
from logging.handlers import RotatingFileHandler
import openai
from openai.error import APIError, RateLimitError, ServiceUnavailableError
from dotenv import load_dotenv
import os
from flask_cors import CORS
import requests
import threading
from werkzeug.middleware.proxy_fix import ProxyFix
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address
 
# 환경 변수 로드
load_dotenv()
 
# Flask 앱 초기화
app = Flask(__name__)
app.wsgi_app = ProxyFix(app.wsgi_app, x_for=1, x_proto=1, x_host=1, x_prefix=1)
CORS(app)
 
# 요청 제한 설정
limiter = Limiter(
    get_remote_address,
    app=app,
    default_limits=["200 per day", "50 per hour"]
)
 
# 로깅 설정
if not os.path.exists('logs'):
    os.mkdir('logs')
file_handler = RotatingFileHandler('logs/app.log', maxBytes=10240, backupCount=10)
file_handler.setFormatter(logging.Formatter(
    '%(asctime)s %(levelname)s: %(message)s [in %(pathname)s:%(lineno)d]'
))
file_handler.setLevel(logging.INFO)
app.logger.addHandler(file_handler)
app.logger.setLevel(logging.INFO)
app.logger.info('애플리케이션 시작')
 
# OpenAI API 키 설정
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
if not OPENAI_API_KEY:
    raise ValueError("OPENAI_API_KEY 환경 변수가 설정되지 않았다.")
openai.api_key = OPENAI_API_KEY
 
# 시크릿 키 설정
SECRET_KEY = os.getenv("SECRET_KEY")
if not SECRET_KEY:
    raise ValueError("SECRET_KEY 환경 변수가 설정되지 않았다.")
app.config['SECRET_KEY'] = SECRET_KEY
 
@app.route('/')
def home():
    return "서버가 돌아가고 있다!"
 
# 경상도 사투리 프롬프트
GYEONGSANG_PROMPT = """
니는 부산 사투리 쓰는 40대 아재다. 다음 규칙 지켜라:
반드시 부산에 오래 산 사람이 사용할 만한 어휘를 많이 사용하고, 반말로 대답해라.
   - 단디: 꼭, 잘
   - 만다꼬?: 왜 그래?
   - 고마 쌔리 마!: 그만해!
   - 우야꼬: 어떡하지?
   - 내사 괘안타: 난 괜찮아
   - 밥 문나?: 밥 먹었어?
   - 되게: 매우
   - 아이가: ~이잖아
   - 뭇노: 먹어
   - 카더라: ~라고 하더라
   - 와그라노: 왜 그러냐
 
"""
 
# GPT와의 대화를 처리하는 함수
def chat_with_gpt(messages):
    try:
        response = openai.ChatCompletion.create(
            model="gpt-4",
            messages=messages
        )
        return response.choices[0].message['content']
    except (APIError, RateLimitError, ServiceUnavailableError) as e:
        app.logger.error(f"OpenAI API 오류: {str(e)}")
        return "아이고, 서비스에 문제가 생겼다. 쪼매 있다 다시 해봐라."
    except Exception as e:
        app.logger.error(f"GPT 채팅 중 오류 발생: {str(e)}")
        return "아이구야, 뭐가 잘못됐는지 모르겠다. 다시 한 번 물어봐라."
 
# 비동기로 GPT 응답을 처리하고 콜백을 보내는 함수
def process_and_callback(user_message, callback_url):
    messages = [
        {"role": "system", "content": GYEONGSANG_PROMPT},
        {"role": "user", "content": user_message}
    ]
    gpt_response = chat_with_gpt(messages)
 
    callback_response = {
        "version": "2.0",
        "template": {
            "outputs": [
                {
                    "simpleText": {
                        "text": gpt_response
                    }
                }
            ]
        }
    }
 
    try:
        response = requests.post(callback_url, json=callback_response, timeout=5)
        app.logger.info(f"콜백 응답 상태 코드: {response.status_code}")
        app.logger.info(f"콜백 응답 내용: {response.text}")
    except requests.exceptions.RequestException as e:
        app.logger.error(f"콜백 전송 중 오류 발생: {str(e)}")
 
# 채팅 엔드포인트
@app.route('/chat', methods=['POST'])
@limiter.limit("10 per minute")
def chat():
    data = request.get_json()
    user_message = data.get('userRequest', {}).get('utterance')
    callback_url = data.get('userRequest', {}).get('callbackUrl')
    
    if not user_message:
        return jsonify({
            "version": "2.0",
            "template": {
                "outputs": [
                    {
                        "simpleText": {
                            "text": "뭐라 카는지 모르겠다. 다시 말해봐라."
                        }
                    }
                ]
            }
        })
 
    if callback_url:
        # 비동기 처리 시작
        threading.Thread(target=process_and_callback, args=(user_message, callback_url)).start()
        
        # 즉시 응답
        return jsonify({
            "version": "2.0",
            "useCallback": True,
            "data": {
                "text": "잠깐만 기다려봐라. 답변을 준비하고 있다."
            }
        })
    else:
        # 콜백 URL이 없는 경우 동기적으로 처리
        messages = [
            {"role": "system", "content": GYEONGSANG_PROMPT},
            {"role": "user", "content": user_message}
        ]
        gpt_response = chat_with_gpt(messages)
 
        return jsonify({
            "version": "2.0",
            "template": {
                "outputs": [
                    {
                        "simpleText": {
                            "text": gpt_response
                        }
                    }
                ]
            }
        })
 
if __name__ == "__main__":
    app.run(host='0.0.0.0', port=5000)  # HTTP로 실행

이제 각 부분에 대해 자세히 설명하겠습니다:

라이브러리 임포트

from flask import Flask, request, jsonify
import logging
from logging.handlers import RotatingFileHandler
import openai
from openai.error import APIError, RateLimitError, ServiceUnavailableError
from dotenv import load_dotenv
import os
from flask_cors import CORS
import requests
import threading
from werkzeug.middleware.proxy_fix import ProxyFix
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address

이 부분에서는 필요한 모든 라이브러리를 임포트합니다. 주요 라이브러리는 다음과 같습니다:

  • Flask: 웹 애플리케이션 프레임워크
  • logging: 로그 기록을 위한 모듈
  • openai: OpenAI API 사용을 위한 라이브러리
  • dotenv: 환경 변수 로드를 위한 라이브러리
  • CORS: Cross-Origin Resource Sharing 설정을 위한 Flask 확장
  • flask_limiter: API 요청 제한을 위한 Flask 확장

환경 변수 로드

load_dotenv()

이 라인은 .env 파일에서 환경 변수를 로드합니다.

Flask 앱 초기화

app = Flask(__name__)
app.wsgi_app = ProxyFix(app.wsgi_app, x_for=1, x_proto=1, x_host=1, x_prefix=1)
CORS(app)

Flask 애플리케이션을 초기화하고, ProxyFix를 적용하여 프록시 서버 뒤에서 실행될 때 올바른 클라이언트 IP를 얻을 수 있도록 합니다. CORS 설정으로 크로스 오리진 요청을 허용합니다.

요청 제한 설정

limiter = Limiter(
    get_remote_address,
    app=app,
    default_limits=["200 per day", "50 per hour"]
)

Flask-Limiter를 사용하여 API 요청을 제한합니다. 기본적으로 하루 200회, 시간당 50회로 제한됩니다.

로깅 설정

if not os.path.exists('logs'):
    os.mkdir('logs')
file_handler = RotatingFileHandler('logs/app.log', maxBytes=10240, backupCount=10)
file_handler.setFormatter(logging.Formatter(
    '%(asctime)s %(levelname)s: %(message)s [in %(pathname)s:%(lineno)d]'
))
file_handler.setLevel(logging.INFO)
app.logger.addHandler(file_handler)
app.logger.setLevel(logging.INFO)
app.logger.info('애플리케이션 시작')

로그 파일을 생성하고 관리하기 위한 설정입니다. RotatingFileHandler를 사용하여 로그 파일의 크기를 제한하고, 백업 파일을 생성합니다.

API 키 및 시크릿 키 설정

OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
if not OPENAI_API_KEY:
    raise ValueError("OPENAI_API_KEY 환경 변수가 설정되지 않았습니다.")
openai.api_key = OPENAI_API_KEY
 
SECRET_KEY = os.getenv("SECRET_KEY")
if not SECRET_KEY:
    raise ValueError("SECRET_KEY 환경 변수가 설정되지 않았습니다.")
app.config['SECRET_KEY'] = SECRET_KEY

환경 변수에서 OpenAI API 키와 Flask 애플리케이션의 시크릿 키를 가져옵니다. 키가 설정되지 않은 경우 오류를 발생시킵니다.

라우트 설정

@app.route('/')
def home():
    return "서버가 실행 중입니다!"

루트 URL(’/‘)에 대한 간단한 응답을 설정합니다.

애플리케이션 실행

if __name__ == "__main__":
    app.run(host='0.0.0.0', port=5000)

애플리케이션을 실행합니다. host='0.0.0.0'은 외부에서 접근 가능하도록 설정하며, port=5000은 5000번 포트를 사용하도록 지정합니다.

이 기본 설정으로 Flask 애플리케이션이 준비되었습니다. 이후 추가적인 라우트와 기능을 구현하여 챗봇 서비스를 완성할 수 있습니다.

Graph View

Backlinks

  • No backlinks found