아래의 Flask 애플리케이션 코드는 Qdrant 벡터 데이터베이스와 OpenAI API를 연동하여 간단한 RAG(Retrieval Augmented Generation) 기반 챗봇 서비스를 구현한 것입니다. 각 부분에 대한 상세 정리와 함께, 더 좋은 코드가 되기 위한 개선 사항 및 추가 코드를 정리합니다.
1. 코드 전체 해설
이 코드는 크게 세 가지 주요 구성 요소로 나뉩니다: Flask 웹 서버, Qdrant 벡터 데이터베이스 클라이언트, OpenAI API 클라이언트.
1.1. 초기 설정 및 클라이언트 초기화
from flask import Flask, request, jsonify # Flask 웹 프레임워크 관련 모듈
from qdrant_client import QdrantClient # Qdrant 서버와 통신하는 클라이언트
from qdrant_client.models import Distance, VectorParams, PointStruct # Qdrant 데이터 모델 정의
import uuid # 고유 ID 생성을 위한 모듈 (UUID)
import openai # OpenAI API와 통신하는 클라이언트
from dotenv import load_dotenv # .env 파일에서 환경 변수를 로드하는 모듈
import os # 운영체제 환경 변수에 접근하는 모듈
# .env 파일 로드: 프로젝트 루트 디렉토리에 있는 .env 파일에서 환경 변수를 읽어옵니다.
load_dotenv()
# OpenAI API 키 가져오기: 환경 변수에서 OPENAI_API_KEY를 가져옵니다.
openai_api_key = os.getenv("OPENAI_API_KEY")
# API 키 유효성 검사: 만약 API 키가 설정되지 않았다면 ValueError를 발생시켜 앱 시작을 중단합니다.
if not openai_api_key:
raise ValueError("OPENAI_API_KEY 환경 변수가 설정되지 않았습니다.")
# OpenAI 클라이언트 초기화: 가져온 API 키를 사용하여 OpenAI 서비스에 접근할 수 있도록 클라이언트를 설정합니다.
openai_client = openai.OpenAI(api_key = openai_api_key)
# Qdrant 클라이언트 초기화: Qdrant 서버에 연결합니다.
# host="localhost": Flask 앱과 Qdrant Docker 컨테이너가 같은 EC2 인스턴스에서 실행될 때 사용합니다.
# port=6333: Qdrant의 기본 gRPC 통신 포트입니다.
qdrant_client = QdrantClient(host="localhost", port=6333)
# Qdrant 서버 연결 테스트: 서버에 연결하여 현재 존재하는 컬렉션 목록을 가져와 성공 여부를 출력합니다.
try:
collections = qdrant_client.get_collections()
print("Qdrant 클라이언트 연결 및 컬렉션 정보 가져오기 성공: ", collections)
except Exception as e:
print(f"Qdrant 클라이언트 연결 또는 작업 중 오류 발생: {e}")
# Flask 애플리케이션 초기화: 웹 서버를 생성합니다.
app = Flask(__name__)
해설:
- 필요한 라이브러리들을 임포트합니다.
- .env 파일에서 OPENAI_API_KEY를 로드하고, 유효성을 검사하여 OpenAI 클라이언트를 초기화합니다.
- localhost:6333으로 Qdrant 클라이언트를 초기화하고, Qdrant 서버와의 연결을 테스트합니다. 이는 앱이 시작될 때 Qdrant 연결 상태를 확인하는 좋은 방법입니다.
1.2. Flask 라우트 정의
@app.route("/")
def home():
return "Welcome to the Flask App!"
@app.route("/test")
def test():
return "test"
해설:
- @app.route("/"): 웹 서버의 루트 경로(예: http://localhost:5500/)에 대한 GET 요청을 처리하는 home 함수를 정의합니다.
- @app.route("/test"): /test 경로(예: http://localhost:5500/test)에 대한 GET 요청을 처리하는 test 함수를 정의합니다.
1.3. OpenAI 챗봇 API 엔드포인트 (/openaiChat)
@app.route("/openaiChat", methods=["POST"])
def chat():
data = request.json # POST 요청의 JSON 본문을 파싱합니다.
question = data.get("question") # JSON에서 'question' 필드 값을 가져옵니다.
prompt = data.get("prompt") # JSON에서 'prompt' 필드 값을 가져옵니다.
# OpenAI 챗봇 API 호출: gpt-4o-mini 모델을 사용하여 대화를 생성합니다.
response = openai_client.chat.completions.create(
model = "gpt-4o-mini", # 사용할 챗봇 모델 지정
messages =[ # 대화 기록 (시스템 프롬프트와 사용자 질문)
{"role": "system", "content": prompt}, # 시스템의 역할과 지시사항
{"role": "user", "content": question}, # 사용자의 질문
],
temperature = 0.8, # 창의성 조절 (0.0은 보수적, 1.0은 창의적)
max_tokens = 300, # 생성될 답변의 최대 토큰 수
)
# 생성된 답변 반환: OpenAI 응답 객체에서 첫 번째 선택지의 메시지 내용을 추출하여 클라이언트에 반환합니다.
return response.choices[0].message.content
해설:
- 클라이언트로부터 question과 prompt를 받아 OpenAI gpt-4o-mini 모델로 챗봇 응답을 생성합니다.
- messages 배열에 시스템 프롬프트와 사용자 질문을 넣어 대화 컨텍스트를 제공합니다.
- temperature와 max_tokens로 답변의 특성을 조절합니다.
1.4. Qdrant 포인트 삽입 엔드포인트 (/upsertPoint)
@app.route("/upsertPoint", methods=["POST"])
def upsertPoint():
data = request.json # POST 요청의 JSON 본문을 파싱합니다.
inputText = data.get("inputText") # JSON에서 'inputText' 필드 값을 가져옵니다.
# OpenAI 임베딩 API 호출: 입력 텍스트를 벡터로 변환합니다.
response = openai_client.embeddings.create(
input = inputText, # 임베딩할 텍스트
model = "text-embedding-3-small", # 1536차원 벡터를 생성하는 임베딩 모델
)
# PointStruct 생성: Qdrant에 저장할 포인트 객체를 만듭니다.
points = [
PointStruct(
id = str(uuid.uuid4()), # 고유한 UUID를 ID로 사용합니다. (중복 방지)
vector = response.data[0].embedding, # OpenAI에서 생성된 1536차원 벡터
payload = {"name": inputText} # 원본 텍스트를 메타데이터(payload)로 저장합니다.
)
]
# Qdrant에 포인트 삽입/업데이트: 'test3' 컬렉션에 정의된 포인트를 upsert합니다.
qdrant_response = qdrant_client.upsert(
collection_name = "test3", # 대상 컬렉션 이름
points = points, # 삽입할 포인트 리스트
)
# print(qdrant_response) # 디버깅을 위해 응답을 출력할 수 있습니다.
return "OK" # 성공 시 'OK' 문자열 반환
해설:
- 클라이언트로부터 inputText를 받아 OpenAI text-embedding-3-small 모델로 임베딩 벡터를 생성합니다.
- 생성된 벡터와 원본 텍스트를 포함하는 PointStruct를 생성합니다. uuid.uuid4()를 사용하여 고유한 ID를 부여하는 것은 매우 좋은 방법입니다.
- test3 컬렉션에 이 포인트를 upsert합니다.
1.5. Qdrant 포인트 검색 엔드포인트 (/queryPoint)
@app.route("/queryPoint", methods=["POST"])
def queryPoint():
data = request.json # POST 요청의 JSON 본문을 파싱합니다.
inputText = data.get("inputText") # JSON에서 검색할 질의 텍스트 'inputText'를 가져옵니다.
# OpenAI 임베딩 API 호출: 질의 텍스트를 벡터로 변환합니다.
response = openai_client.embeddings.create(
input = inputText, # 임베딩할 질의 텍스트
model = "text-embedding-3-small", # 데이터 삽입 시 사용한 모델과 동일해야 합니다.
)
# Qdrant에서 유사한 포인트 검색:
search_response = qdrant_client.query_points(
collection_name = "test3", # 검색 대상 컬렉션
query = response.data[0].embedding, # 변환된 질의 벡터
limit = 3, # 가장 유사한 상위 3개 포인트 반환
with_payload = True, # 검색 결과에 페이로드 포함
)
# 검색된 포인트들의 페이로드 추출:
payloads = [point.payload for point in search_response.points]
print(payloads) # 추출된 페이로드들을 서버 콘솔에 출력 (디버깅용)
return payloads # 추출된 페이로드 리스트를 JSON 형태로 클라이언트에 반환
해설:
- 클라이언트로부터 inputText를 받아 OpenAI 임베딩 모델로 쿼리 벡터를 생성합니다.
- 생성된 쿼리 벡터를 사용하여 test3 컬렉션에서 가장 유사한 상위 3개의 포인트를 검색합니다.
- 검색된 포인트들에서 payload (메타데이터)만 추출하여 클라이언트에 반환합니다.
1.6. RAG 기반 챗봇 엔드포인트 (/searchChat)
@app.route("/searchChat", methods =['POST'])
def searchChat():
data = request.json # POST 요청의 JSON 본문을 파싱합니다.
inputText = data.get("inputText") # JSON에서 사용자 질문 'inputText'를 가져옵니다.
# OpenAI 임베딩 API 호출: 사용자 질문을 벡터로 변환합니다.
response = openai_client.embeddings.create(
input = inputText,
model = "text-embedding-3-small",
)
# Qdrant에서 관련 정보 검색: 사용자 질문 벡터와 유사한 상위 3개 포인트를 'test3' 컬렉션에서 검색합니다.
search_response = qdrant_client.query_points(
collection_name ="test3",
query = response.data[0].embedding,
limit=3,
with_payload = True, # 페이로드 포함
)
# 검색된 포인트들의 페이로드(참조 정보) 추출:
payloads = [point.payload for point in search_response.points]
# OpenAI 챗봇 API 호출 (RAG): 검색된 참조 정보를 기반으로 답변을 생성합니다.
reference_response = openai_client.chat.completions.create(
model = "gpt-4o-mini", # 챗봇 모델
messages = [
# 시스템 프롬프트: 챗봇의 역할과 답변 생성 지침을 명시합니다.
{"role": "system", "content":"user 질문에 대해서, Reference를 기반으로 답변해요."},
# 사용자 프롬프트: 검색된 참조 정보와 실제 사용자 질문을 결합하여 LLM에 전달합니다.
{"role":"user", "content":"#Reference" + "\n".join(p["name"] for p in payloads)
+ "user 질문" + inputText}, # p["name"]은 페이로드의 'name' 필드 값을 가져옵니다.
],
temperature = 0.8,
max_tokens = 300,
)
# 생성된 답변 반환: LLM이 생성한 답변을 클라이언트에 반환합니다.
return reference_response.choices[0].message.content
해설:
- 사용자 질문(inputText)을 받아 OpenAI 임베딩 모델로 벡터화합니다.
- 이 벡터를 사용하여 Qdrant에서 관련성이 높은 정보를 검색합니다 (test3 컬렉션에서 상위 3개).
- 검색된 정보(페이로드의 name 필드)를 #Reference 섹션에 넣어 OpenAI 챗봇 모델의 프롬프트로 전달합니다.
- 챗봇은 이 참조 정보를 바탕으로 사용자 질문에 대한 답변을 생성하고 반환합니다. 이것이 바로 RAG의 핵심 원리입니다.
1.7. 앱 실행 (if __name__ == "__main__":)
if __name__ == "__main__":
app.run(debug=True, port=5500)
해설:
- 이 부분은 스크립트가 직접 실행될 때만 app.run()을 호출하도록 합니다.
- debug=True: 개발 모드를 활성화합니다. 코드 변경 시 자동으로 서버를 재시작하고, 상세한 에러 메시지를 제공합니다. (프로덕션 환경에서는 False로 설정해야 합니다.)
- port=5500: Flask 앱이 5500번 포트에서 리스닝하도록 설정합니다.
2. 더 좋은 코드가 되기 위한 개선 사항 및 추가 코드
현재 코드는 기본적인 기능을 잘 수행하고 있지만, 실제 서비스 환경이나 더 견고한 애플리케이션을 위해서는 몇 가지 개선 사항을 고려할 수 있습니다.
2.1. CORS 설정 추가
프론트엔드(React 앱)가 다른 포트나 도메인에서 Flask 백엔드에 접근할 경우, CORS(Cross-Origin Resource Sharing) 문제가 발생할 수 있습니다.
추가 코드:
# ... (기존 임포트) ...
from flask_cors import CORS # Flask-CORS 라이브러리 임포트
# ... (Qdrant 클라이언트 초기화 아래) ...
app = Flask(__name__)
CORS(app) # Flask 앱에 CORS를 적용합니다.
# 또는 특정 도메인만 허용하려면: CORS(app, resources={r"/*": {"origins": "http://your-frontend-domain.com"}})
설명:
- pip install Flask-Cors로 라이브러리를 설치해야 합니다.
- CORS(app)를 추가하면 모든 도메인에서의 요청을 허용합니다. 보안을 위해 특정 프론트엔드 도메인만 허용하도록 설정하는 것이 좋습니다.
2.2. 에러 핸들링 강화 및 응답 통일
현재는 Flask 내부에서 에러가 발생하면 500 에러가 클라이언트에 그대로 전달됩니다. 사용자에게 더 친화적인 에러 메시지를 제공하고, 모든 API 응답 형식을 통일하는 것이 좋습니다.
추가 코드 (예시):
# ... (app = Flask(__name__) 아래) ...
# 모든 API 응답을 JSON 형태로 통일하기 위한 헬퍼 함수
def create_response(data, status_code=200, message="Success"):
return jsonify({
"status": "success" if 200 <= status_code < 300 else "error",
"message": message,
"data": data
}), status_code
# 전역 에러 핸들러 (예시: 500 Internal Server Error)
@app.errorhandler(500)
def handle_internal_server_error(e):
# 실제 에러 로깅 (예: Sentry, CloudWatch Logs)
print(f"Server Error: {e}")
return create_response(None, 500, "서버 내부 오류가 발생했습니다. 잠시 후 다시 시도해주세요.")
# 특정 엔드포인트의 에러 핸들링 예시 (try-except 블록 활용)
@app.route("/upsertPoint", methods=["POST"])
def upsertPoint():
try:
data = request.json
inputText = data.get("inputText")
if not inputText:
return create_response(None, 400, "inputText가 필요합니다.")
response = openai_client.embeddings.create(
input = inputText,
model = "text-embedding-3-small",
)
points = [
PointStruct(
id = str(uuid.uuid4()),
vector = response.data[0].embedding,
payload = {"name": inputText}
)
]
qdrant_response = qdrant_client.upsert(
collection_name = "test3",
points = points,
)
return create_response({"qdrant_status": qdrant_response.status.ok}, 200, "포인트가 성공적으로 삽입되었습니다.")
except openai.APIError as e:
print(f"OpenAI API Error: {e}")
return create_response(None, 500, f"OpenAI API 오류: {e.code}")
except Exception as e:
print(f"Error in upsertPoint: {e}")
return create_response(None, 500, "포인트 삽입 중 오류가 발생했습니다.")
# 다른 엔드포인트들도 유사하게 try-except 블록과 create_response 함수를 사용하여 에러를 처리하고 응답을 통일할 수 있습니다.
설명:
- create_response 함수를 통해 모든 API 응답을 {"status": "success/error", "message": "", "data": {}}와 같은 일관된 JSON 형식으로 만듭니다.
- @app.errorhandler(500)을 사용하여 서버 내부 오류 발생 시 사용자에게 친화적인 메시지를 반환하도록 합니다.
- 각 엔드포인트 내부에 try-except 블록을 추가하여 OpenAI API 오류, Qdrant 클라이언트 오류 등 특정 예외를 처리하고 적절한 HTTP 상태 코드와 메시지를 반환합니다.
2.3. 비동기 처리 (선택 사항, 성능 개선)
현재 Flask 앱은 동기적으로 요청을 처리합니다. OpenAI API 호출이나 Qdrant 작업은 네트워크 I/O를 포함하므로 시간이 걸릴 수 있습니다. 많은 요청이 동시에 들어올 경우 서버가 블로킹될 수 있습니다. asyncio와 httpx (Qdrant 클라이언트가 내부적으로 사용)를 활용하여 비동기 처리로 전환하면 성능을 향상시킬 수 있습니다.
추가 코드 (예시, Flask-Asyncio 등 필요):
# ... (기존 임포트) ...
# from flask_cors import CORS # Flask-CORS를 사용한다면 임포트
# from flask_asyncio import FlaskAsyncIO # pip install Flask-AsyncIO
# app = FlaskAsyncIO(__name__) # FlaskAsyncIO로 변경
# CORS(app) # CORS 사용 시
# async def chat(): # async 함수로 변경
# # ... (내부 로직은 거의 동일) ...
# response = await openai_client.chat.completions.create(...) # await 추가
# return response.choices[0].message.content
# if __name__ == "__main__":
# app.run(debug=True, port=5500)
설명:
- Flask-AsyncIO와 같은 라이브러리를 사용하거나, gunicorn과 같은 비동기 WSGI 서버를 gevent나 eventlet 워커와 함께 사용하는 것을 고려해야 합니다.
- 각 엔드포인트 함수를 async def로 정의하고, 네트워크 I/O가 발생하는 부분(OpenAI, Qdrant 호출)에 await를 붙여 비동기적으로 실행되도록 합니다.
2.4. 로깅 (Logging)
현재 print() 문으로 로그를 출력하고 있지만, 실제 애플리케이션에서는 Python의 logging 모듈을 사용하여 체계적인 로깅을 구현하는 것이 좋습니다.
추가 코드 (예시):
import logging
# 로거 설정
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
# ... (기존 코드) ...
# Qdrant 클라이언트 연결 테스트 부분
try:
collections = qdrant_client.get_collections()
logger.info(f"Qdrant 클라이언트 연결 및 컬렉션 정보 가져오기 성공: {collections}")
except Exception as e:
logger.error(f"Qdrant 클라이언트 연결 또는 작업 중 오류 발생: {e}")
# ... (각 엔드포인트 내에서도 print 대신 logger.info, logger.error 사용) ...
설명:
- logging 모듈을 설정하여 정보, 경고, 에러 메시지를 체계적으로 기록합니다.
- 이는 문제 발생 시 원인 파악과 디버깅에 큰 도움이 됩니다.
2.5. 설정 외부화
collection_name이나 model 이름 같은 설정 값들을 코드 내에 하드코딩하기보다는, 환경 변수나 별도의 설정 파일로 관리하는 것이 좋습니다.
추가 코드 (예시):
# ... (기존 임포트) ...
# .env 파일에서 Qdrant 컬렉션 이름 가져오기
QDRANT_COLLECTION_NAME = os.getenv("QDRANT_COLLECTION_NAME", "test3") # 기본값은 'test3'
OPENAI_EMBEDDING_MODEL = os.getenv("OPENAI_EMBEDDING_MODEL", "text-embedding-3-small")
OPENAI_CHAT_MODEL = os.getenv("OPENAI_CHAT_MODEL", "gpt-4o-mini")
# ... (각 엔드포인트에서 사용) ...
# qdrant_client.upsert(collection_name = QDRANT_COLLECTION_NAME, ...)
# openai_client.embeddings.create(model = OPENAI_EMBEDDING_MODEL, ...)
# openai_client.chat.completions.create(model = OPENAI_CHAT_MODEL, ...)
설명:
- .env 파일에 QDRANT_COLLECTION_NAME=my_rag_collection과 같이 설정하고 코드에서 가져와 사용합니다.
- 코드를 더 유연하게 만들고, 환경에 따라 쉽게 변경할 수 있습니다.
최종 요약:
위 코드는 Flask, Qdrant, OpenAI를 연동하는 기본적인 RAG 시스템의 핵심 로직을 구현하고 있습니다. 추가로 위에서 제안된 개선 사항들은 주로 견고성, 유지보수성, 보안, 성능 측면을 강화하기 위한 것으로, 실제 프로덕션 환경에서는 매우 중요하게 고려되어야 할 부분들입니다. 개발 초기 단계에서는 현재 코드로 충분히 테스트하고 기능을 구현할 수 있습니다.
3. 기본 코드 + 개선 사항 코드
from flask import Flask, request, jsonify
from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct
import uuid
import openai
from dotenv import load_dotenv
import os
from flask_cors import CORS # CORS를 위한 Flask-CORS 임포트
import logging # 로깅을 위한 logging 모듈 임포트
# --- 1. 초기 설정 및 클라이언트 초기화 ---
# .env 파일 로드
load_dotenv()
# 환경 변수에서 API 키 및 설정 값 가져오기
openai_api_key = os.getenv("OPENAI_API_KEY")
if not openai_api_key:
raise ValueError("OPENAI_API_KEY 환경 변수가 설정되지 않았습니다.")
# Qdrant 컬렉션 및 OpenAI 모델 이름 설정 (환경 변수에서 가져오거나 기본값 사용)
QDRANT_COLLECTION_NAME = os.getenv("QDRANT_COLLECTION_NAME", "test3")
OPENAI_EMBEDDING_MODEL = os.getenv("OPENAI_EMBEDDING_MODEL", "text-embedding-3-small")
OPENAI_CHAT_MODEL = os.getenv("OPENAI_CHAT_MODEL", "gpt-4o-mini")
# 로깅 설정
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
# OpenAI 클라이언트 초기화
openai_client = openai.OpenAI(api_key=openai_api_key)
# Qdrant 클라이언트 초기화 (Flask 앱과 Qdrant Docker 컨테이너가 같은 EC2 인스턴스에서 실행될 때 host="localhost" 사용)
qdrant_client = QdrantClient(host="localhost", port=6333)
# Qdrant 서버 연결 테스트
try:
collections = qdrant_client.get_collections()
logger.info(f"Qdrant 클라이언트 연결 및 컬렉션 정보 가져오기 성공: {collections}")
except Exception as e:
logger.error(f"Qdrant 클라이언트 연결 또는 작업 중 오류 발생: {e}")
# 연결 실패 시 앱 시작을 중단할 수도 있습니다.
# raise ConnectionError(f"Qdrant 클라이언트 연결 실패: {e}")
# Flask 애플리케이션 초기화
app = Flask(__name__)
CORS(app) # 모든 도메인에서의 요청을 허용하는 CORS 설정
# 특정 도메인만 허용하려면: CORS(app, resources={r"/*": {"origins": "http://your-frontend-domain.com"}})
# --- 2. API 응답 통일 및 에러 핸들링 헬퍼 함수 ---
def create_response(data, status_code=200, message="Success"):
"""
모든 API 응답을 일관된 JSON 형식으로 생성하는 헬퍼 함수.
"""
return jsonify({
"status": "success" if 200 <= status_code < 300 else "error",
"message": message,
"data": data
}), status_code
# 전역 에러 핸들러 (500 Internal Server Error)
@app.errorhandler(500)
def handle_internal_server_error(e):
logger.exception("서버 내부 오류 발생:") # 상세한 예외 정보 로깅
return create_response(None, 500, "서버 내부 오류가 발생했습니다. 잠시 후 다시 시도해주세요.")
# --- 3. Flask 라우트 정의 ---
@app.route("/")
def home():
"""루트 경로에 대한 GET 요청 처리."""
logger.info("Home 페이지 요청 수신")
return "Welcome to the Flask App!"
@app.route("/test")
def test():
"""테스트 경로에 대한 GET 요청 처리."""
logger.info("Test 페이지 요청 수신")
return "test"
@app.route("/openaiChat", methods=["POST"])
def chat():
"""
OpenAI 챗봇 API 엔드포인트.
사용자 질문과 시스템 프롬프트를 받아 챗봇 응답을 생성합니다.
"""
try:
data = request.json
question = data.get("question")
prompt = data.get("prompt")
if not question:
return create_response(None, 400, "질문(question)이 필요합니다.")
logger.info(f"OpenAI Chat 요청 수신 - 질문: {question}")
response = openai_client.chat.completions.create(
model=OPENAI_CHAT_MODEL,
messages=[
{"role": "system", "content": prompt if prompt else "당신은 친절한 챗봇입니다."},
{"role": "user", "content": question},
],
temperature=0.8,
max_tokens=300,
)
chat_response_content = response.choices[0].message.content
logger.info(f"OpenAI Chat 응답 생성 완료: {chat_response_content[:50]}...") # 응답 일부 로깅
return create_response(chat_response_content, 200, "챗봇 응답 생성 완료")
except openai.APIError as e:
logger.error(f"OpenAI API 오류 발생 (/openaiChat): {e.code} - {e.response.text}")
return create_response(None, 500, f"OpenAI API 오류: {e.code}")
except Exception as e:
logger.exception("오류 발생 (/openaiChat):") # 예외 발생 시 상세 로깅
return create_response(None, 500, "챗봇 응답 생성 중 오류가 발생했습니다.")
@app.route("/upsertPoint", methods=["POST"])
def upsertPoint():
"""
Qdrant에 포인트(벡터 및 페이로드)를 삽입하는 엔드포인트.
입력 텍스트를 임베딩하여 Qdrant 컬렉션에 저장합니다.
"""
try:
data = request.json
inputText = data.get("inputText")
if not inputText:
return create_response(None, 400, "inputText가 필요합니다.")
logger.info(f"UpsertPoint 요청 수신 - 입력 텍스트: {inputText}")
# OpenAI 임베딩 API 호출
response = openai_client.embeddings.create(
input=inputText,
model=OPENAI_EMBEDDING_MODEL,
)
# PointStruct 생성 (UUID로 고유 ID 부여)
points = [
PointStruct(
id=str(uuid.uuid4()),
vector=response.data[0].embedding,
payload={"name": inputText}
)
]
# Qdrant에 포인트 삽입/업데이트
qdrant_response = qdrant_client.upsert(
collection_name=QDRANT_COLLECTION_NAME,
points=points,
)
logger.info(f"Qdrant Upsert 완료: {qdrant_response.status.ok}")
return create_response({"qdrant_status": qdrant_response.status.ok}, 200, "포인트가 성공적으로 삽입되었습니다.")
except openai.APIError as e:
logger.error(f"OpenAI API 오류 발생 (/upsertPoint): {e.code} - {e.response.text}")
return create_response(None, 500, f"OpenAI API 오류: {e.code}")
except Exception as e:
logger.exception("오류 발생 (/upsertPoint):")
return create_response(None, 500, "포인트 삽입 중 오류가 발생했습니다.")
@app.route("/queryPoint", methods=["POST"])
def queryPoint():
"""
Qdrant에서 유사한 포인트(벡터)를 검색하는 엔드포인트.
입력 텍스트를 임베딩하여 Qdrant 컬렉션에서 유사한 데이터를 찾습니다.
"""
try:
data = request.json
inputText = data.get("inputText")
if not inputText:
return create_response(None, 400, "inputText가 필요합니다.")
logger.info(f"QueryPoint 요청 수신 - 질의 텍스트: {inputText}")
# OpenAI 임베딩 API 호출
response = openai_client.embeddings.create(
input=inputText,
model=OPENAI_EMBEDDING_MODEL,
)
# Qdrant에서 유사한 포인트 검색 (query_points 메서드 사용)
search_response = qdrant_client.query_points(
collection_name=QDRANT_COLLECTION_NAME,
query=response.data[0].embedding,
limit=3,
with_payload=True, # 페이로드 포함
)
# 검색된 포인트들의 페이로드 추출
payloads = [point.payload for point in search_response.points]
logger.info(f"Qdrant 검색 완료 - 결과 페이로드: {payloads}")
return create_response(payloads, 200, "포인트 검색 완료")
except openai.APIError as e:
logger.error(f"OpenAI API 오류 발생 (/queryPoint): {e.code} - {e.response.text}")
return create_response(None, 500, f"OpenAI API 오류: {e.code}")
except Exception as e:
logger.exception("오류 발생 (/queryPoint):")
return create_response(None, 500, "포인트 검색 중 오류가 발생했습니다.")
@app.route("/searchChat", methods=['POST'])
def searchChat():
"""
RAG(Retrieval Augmented Generation) 기반 챗봇 엔드포인트.
사용자 질문을 기반으로 Qdrant에서 관련 정보를 검색하고, 이를 바탕으로 OpenAI 챗봇 응답을 생성합니다.
"""
try:
data = request.json
inputText = data.get("inputText")
if not inputText:
return create_response(None, 400, "inputText가 필요합니다.")
logger.info(f"SearchChat 요청 수신 - 사용자 질문: {inputText}")
# 1. 사용자 질문을 임베딩하여 쿼리 벡터 생성
response = openai_client.embeddings.create(
input=inputText,
model=OPENAI_EMBEDDING_MODEL,
)
# 2. Qdrant에서 관련 정보 검색
search_response = qdrant_client.query_points(
collection_name=QDRANT_COLLECTION_NAME,
query=response.data[0].embedding,
limit=3,
with_payload=True,
)
# 3. 검색된 포인트들의 페이로드(참조 정보) 추출
payloads = [point.payload for point in search_response.points]
logger.info(f"Qdrant에서 검색된 참조 정보: {payloads}")
# 4. OpenAI 챗봇 API 호출 (RAG)
# 검색된 참조 정보를 바탕으로 답변을 생성하도록 프롬프트 구성
reference_content = "\n".join(p["name"] for p in payloads)
user_message_content = f"#Reference\n{reference_content}\n사용자의 질문: {inputText}"
reference_response = openai_client.chat.completions.create(
model=OPENAI_CHAT_MODEL,
messages=[
{"role": "system", "content": "user 질문에 대해서, Reference를 기반으로 답변해요. Reference에 없는 내용은 답변하지 마세요."}, # 시스템 프롬프트 강화
{"role": "user", "content": user_message_content},
],
temperature=0.8,
max_tokens=300,
)
chat_response_content = reference_response.choices[0].message.content
logger.info(f"RAG 챗봇 응답 생성 완료: {chat_response_content[:50]}...")
return create_response(chat_response_content, 200, "RAG 챗봇 응답 생성 완료")
except openai.APIError as e:
logger.error(f"OpenAI API 오류 발생 (/searchChat): {e.code} - {e.response.text}")
return create_response(None, 500, f"OpenAI API 오류: {e.code}")
except Exception as e:
logger.exception("오류 발생 (/searchChat):")
return create_response(None, 500, "RAG 챗봇 응답 생성 중 오류가 발생했습니다.")
# --- 4. 앱 실행 ---
if __name__ == "__main__":
# 프로덕션 환경에서는 debug=False로 설정하고, Gunicorn과 같은 WSGI 서버를 사용하는 것이 좋습니다.
app.run(debug=True, port=5500)
반응형
'AI, 클라우드, 협업, 교육, 문서, 업무자동화' 카테고리의 다른 글
| 벡터DB, Qdrant를 활용한 벡터 검색 워크플로우 정리 (0) | 2025.07.20 |
|---|---|
| 벡터DB, Qdrant 컬렉션 검색 코드 설명 (0) | 2025.07.20 |
| Qdrant 컬렉션에 포인트(Vector & Payload) 삽입 코드 설명 (0) | 2025.07.20 |
| Qdrant 벡터 데이터베이스에 새로운 컬렉션(Collection)을 생성하는 예시 (0) | 2025.07.20 |
| React 기본 사용법 (0) | 2025.07.14 |
