목차
1. 챗봇의 이해
1.1 챗봇이란?
챗봇은 자연어 처리 기술을 활용하여 사용자와 대화를 나누는 AI 시스템입니다. 사용자의 질문이나 명령에 대응하여 적절한 응답을 제공하며, 인간과 유사한 대화 경험을 제공하는 것을 목표로 합니다.
1.2 챗봇의 종류
규칙 기반 챗봇
- 미리 정의된 규칙과 패턴에 따라 응답을 생성
- 키워드 매칭, 의도 인식 등의 단순한 방법 사용
- 장점: 구현이 간단하고 예측 가능한 응답 생성
- 단점: 다양한 표현을 처리하는 데 한계가 있음
AI 기반 챗봇
- 머신러닝과 딥러닝 모델을 활용한 챗봇
- 대규모 데이터로 학습된 언어 모델 사용
- 장점: 자연스러운 대화, 맥락 이해 가능
- 단점: 구현이 복잡하고 컴퓨팅 자원이 많이 필요함
1.3 챗봇의 주요 구성 요소
- 자연어 이해(NLU): 사용자 입력을 분석하여 의도와 엔티티 추출
- 대화 관리: 대화 흐름을 제어하고 상태 추적
- 자연어 생성(NLG): 적절한 응답 생성
- 지식 베이스: 응답 생성에 필요한 정보 저장
- 통합 인터페이스: 다양한 플랫폼과의 연결 기능
2. 챗봇 개발 준비
2.1 개발 환경 설정
필요한 도구 및 라이브러리
- Python 3.7 이상
- 자연어 처리 라이브러리 (NLTK, spaCy, Transformers 등)
- 웹 프레임워크 (Flask, Django, FastAPI 등)
- 데이터베이스 (SQLite, PostgreSQL, MongoDB 등)
# 기본 환경 설정 예시
pip install python-dotenv flask numpy pandas transformers torch
2.2 필요한 API 키 준비
주요 AI 서비스 제공자:
- OpenAI (GPT API)
- Google (Dialogflow, PaLM API)
- Microsoft Azure (Language Understanding, Bot Framework)
- Amazon (Lex, Comprehend)
# .env 파일 예시
OPENAI_API_KEY=your_openai_api_key
MODEL_NAME=gpt-3.5-turbo
2.3 아키텍처 설계
기본 아키텍처:
- 사용자 인터페이스: 메시지 입력 및 표시
- API 서버: 사용자 요청 처리 및 응답 제공
- 자연어 처리 모듈: 입력 분석 및 의도 파악
- 대화 관리 시스템: 대화 상태 및 흐름 관리
- 응답 생성 모듈: 적절한 답변 생성
- 데이터베이스: 대화 기록 및 사용자 정보 저장
3. 기본 챗봇 구현
3.1 OpenAI API를 활용한 간단한 챗봇
import os
import openai
from dotenv import load_dotenv
# 환경 변수 로드
load_dotenv()
openai.api_key = os.getenv("OPENAI_API_KEY")
def get_chatbot_response(user_input, conversation_history=None):
if conversation_history is None:
conversation_history = []
# 대화 기록에 사용자 입력 추가
conversation_history.append({"role": "user", "content": user_input})
# OpenAI API 호출
response = openai.ChatCompletion.create(
model="gpt-3.5-turbo",
messages=[
{"role": "system", "content": "You are a helpful assistant."},
*conversation_history
]
)
# 응답 추출
assistant_response = response.choices[0].message["content"]
# 대화 기록에 응답 추가
conversation_history.append({"role": "assistant", "content": assistant_response})
return assistant_response, conversation_history
# 사용 예시
conversation = []
while True:
user_input = input("You: ")
if user_input.lower() in ["exit", "quit", "bye"]:
break
response, conversation = get_chatbot_response(user_input, conversation)
print(f"Bot: {response}")
3.2 Flask를 이용한 웹 기반 챗봇 인터페이스
from flask import Flask, request, jsonify, render_template
import os
import openai
from dotenv import load_dotenv
# 환경 변수 로드
load_dotenv()
openai.api_key = os.getenv("OPENAI_API_KEY")
app = Flask(__name__)
# 대화 기록 저장 (실제 서비스에서는 데이터베이스 사용 권장)
conversations = {}
@app.route('/')
def index():
return render_template('index.html')
@app.route('/api/chat', methods=['POST'])
def chat():
data = request.json
user_input = data.get('message', '')
user_id = data.get('user_id', 'default_user')
# 해당 사용자의 대화 기록 가져오기
if user_id not in conversations:
conversations[user_id] = []
# 대화 기록에 사용자 입력 추가
conversations[user_id].append({"role": "user", "content": user_input})
# OpenAI API 호출
response = openai.ChatCompletion.create(
model="gpt-3.5-turbo",
messages=[
{"role": "system", "content": "You are a helpful assistant."},
*conversations[user_id]
]
)
# 응답 추출
assistant_response = response.choices[0].message["content"]
# 대화 기록에 응답 추가
conversations[user_id].append({"role": "assistant", "content": assistant_response})
return jsonify({"response": assistant_response})
if __name__ == '__main__':
app.run(debug=True)
<!-- templates/index.html -->
<!DOCTYPE html>
<html>
<head>
<title>AI 챗봇</title>
<style>
#chat-container {
width: 500px;
height: 600px;
margin: 0 auto;
border: 1px solid #ccc;
padding: 20px;
}
#chat-messages {
height: 500px;
overflow-y: auto;
margin-bottom: 10px;
padding: 10px;
border: 1px solid #eee;
}
.user-message {
background-color: #e3f2fd;
padding: 8px 12px;
border-radius: 12px;
margin: 8px 0;
max-width: 70%;
align-self: flex-end;
margin-left: auto;
}
.bot-message {
background-color: #f1f1f1;
padding: 8px 12px;
border-radius: 12px;
margin: 8px 0;
max-width: 70%;
}
</style>
</head>
<body>
<div id="chat-container">
<h1>AI 챗봇</h1>
<div id="chat-messages"></div>
<div id="chat-input">
<input type="text" id="user-input" placeholder="메시지를 입력하세요..." style="width: 80%;">
<button onclick="sendMessage()">전송</button>
</div>
</div>
<script>
// 사용자 ID 생성 (실제 서비스에서는 인증 시스템 연동)
const userId = 'user_' + Math.random().toString(36).substr(2, 9);
// 엔터 키 이벤트 처리
document.getElementById('user-input').addEventListener('keypress', function(e) {
if (e.key === 'Enter') {
sendMessage();
}
});
function sendMessage() {
const userInput = document.getElementById('user-input').value;
if (!userInput.trim()) return;
// 사용자 메시지 화면에 표시
const chatMessages = document.getElementById('chat-messages');
const userMessageElement = document.createElement('div');
userMessageElement.className = 'user-message';
userMessageElement.textContent = userInput;
chatMessages.appendChild(userMessageElement);
// 입력 필드 초기화
document.getElementById('user-input').value = '';
// 스크롤을 최하단으로 이동
chatMessages.scrollTop = chatMessages.scrollHeight;
// 서버에 메시지 전송
fetch('/api/chat', {
method: 'POST',
headers: {
'Content-Type': 'application/json',
},
body: JSON.stringify({
message: userInput,
user_id: userId
}),
})
.then(response => response.json())
.then(data => {
// 봇 응답 화면에 표시
const botMessageElement = document.createElement('div');
botMessageElement.className = 'bot-message';
botMessageElement.textContent = data.response;
chatMessages.appendChild(botMessageElement);
// 스크롤을 최하단으로 이동
chatMessages.scrollTop = chatMessages.scrollHeight;
})
.catch(error => {
console.error('Error:', error);
const errorElement = document.createElement('div');
errorElement.className = 'bot-message';
errorElement.textContent = '오류가 발생했습니다. 다시 시도해주세요.';
chatMessages.appendChild(errorElement);
});
}
</script>
</body>
</html>
3.3 의도 분류 기능 추가
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.naive_bayes import MultinomialNB
from sklearn.pipeline import Pipeline
import pandas as pd
import numpy as np
import pickle
# 의도 분류 모델 클래스
class IntentClassifier:
def __init__(self):
self.model = Pipeline([
('tfidf', TfidfVectorizer(max_features=1000)),
('clf', MultinomialNB())
])
self.is_trained = False
def train(self, texts, labels):
self.model.fit(texts, labels)
self.is_trained = True
def predict(self, text):
if not self.is_trained:
return "unknown"
# 단일 텍스트를 리스트로 변환
if isinstance(text, str):
text = [text]
predictions = self.model.predict(text)
return predictions[0]
def save_model(self, filepath):
with open(filepath, 'wb') as f:
pickle.dump(self.model, f)
def load_model(self, filepath):
with open(filepath, 'rb') as f:
self.model = pickle.load(f)
self.is_trained = True
# 샘플 데이터로 모델 훈련
def train_intent_classifier():
# 간단한 의도 분류 데이터셋
data = {
'text': [
'안녕하세요', '안녕', '반갑습니다', '처음 뵙겠습니다',
'날씨 어때요?', '오늘 날씨가 어떤가요?', '내일 비 올까요?',
'영화 추천해주세요', '볼만한 영화 있을까요?', '최신 영화가 뭐가 있나요?',
'고맙습니다', '감사합니다', '감사해요',
'종료', '그만', '끝내자'
],
'intent': [
'greeting', 'greeting', 'greeting', 'greeting',
'weather', 'weather', 'weather',
'movie_recommendation', 'movie_recommendation', 'movie_recommendation',
'thanks', 'thanks', 'thanks',
'exit', 'exit', 'exit'
]
}
df = pd.DataFrame(data)
# 모델 학습
classifier = IntentClassifier()
classifier.train(df['text'], df['intent'])
# 모델 저장
classifier.save_model('intent_classifier.pkl')
return classifier
# 의도에 따른 응답 함수
def get_response_by_intent(intent, user_input):
responses = {
'greeting': ['안녕하세요! 무엇을 도와드릴까요?', '반갑습니다! 어떤 도움이 필요하신가요?'],
'weather': ['날씨 정보를 알려드리기 위해 지역을 알려주시겠어요?', '어느 지역의 날씨를 알고 싶으신가요?'],
'movie_recommendation': ['어떤 장르의 영화를 좋아하시나요?', '최근에 인기 있는 영화를 추천해 드릴까요?'],
'thanks': ['천만에요!', '별말씀을요. 더 필요한 것이 있으신가요?'],
'exit': ['대화를 종료합니다. 좋은 하루 되세요!', '또 필요하시면 언제든 불러주세요!'],
'unknown': ['죄송합니다. 이해하지 못했어요. 다른 방식으로 질문해 주시겠어요?']
}
return np.random.choice(responses[intent])
# 통합 예시
def chatbot_with_intent_classification(user_input):
# 모델 불러오기 (없으면 훈련)
try:
classifier = IntentClassifier()
classifier.load_model('intent_classifier.pkl')
except:
classifier = train_intent_classifier()
# 의도 분류
intent = classifier.predict(user_input)
# 의도별 응답 생성
response = get_response_by_intent(intent, user_input)
return response, intent
4. 고급 기능 구현
4.1 다국어 지원
from langdetect import detect
from googletrans import Translator
# 언어 감지 및 번역 클래스
class LanguageProcessor:
def __init__(self):
self.translator = Translator()
self.default_language = 'ko' # 기본 언어는 한국어
def detect_language(self, text):
try:
return detect(text)
except:
return self.default_language
def translate_to_default(self, text, source_language=None):
if not source_language:
source_language = self.detect_language(text)
if source_language == self.default_language:
return text
translation = self.translator.translate(text, src=source_language, dest=self.default_language)
return translation.text
def translate_from_default(self, text, target_language):
if target_language == self.default_language:
return text
translation = self.translator.translate(text, src=self.default_language, dest=target_language)
return translation.text
# 사용 예시
def process_multilingual_input(user_input):
processor = LanguageProcessor()
# 언어 감지
detected_language = processor.detect_language(user_input)
print(f"감지된 언어: {detected_language}")
# 기본 언어로 번역
if detected_language != processor.default_language:
translated_input = processor.translate_to_default(user_input, detected_language)
print(f"번역된 입력: {translated_input}")
else:
translated_input = user_input
# 챗봇 처리 (기본 언어로)
# ... 챗봇 로직 ...
bot_response = f"당신의 메시지 '{translated_input}'를 받았습니다."
# 원래 언어로 번역하여 응답
if detected_language != processor.default_language:
final_response = processor.translate_from_default(bot_response, detected_language)
else:
final_response = bot_response
return final_response, detected_language
4.2 대화 맥락 관리
from datetime import datetime
import json
class ConversationManager:
def __init__(self, max_history_len=10):
self.conversations = {}
self.max_history_len = max_history_len
def add_message(self, user_id, role, content):
# 해당 사용자의 대화 기록이 없으면 초기화
if user_id not in self.conversations:
self.conversations[user_id] = []
# 메시지 추가
self.conversations[user_id].append({
"role": role,
"content": content,
"timestamp": datetime.now().isoformat()
})
# 최대 길이 제한
if len(self.conversations[user_id]) > self.max_history_len * 2: # 사용자와 봇 메시지 쌍으로 계산
# 오래된 메시지부터 제거 (시스템 메시지는 유지)
self.conversations[user_id] = [
msg for msg in self.conversations[user_id]
if msg["role"] == "system"
] + self.conversations[user_id][-(self.max_history_len * 2):]
def get_conversation(self, user_id):
return self.conversations.get(user_id, [])
def get_formatted_messages(self, user_id):
"""OpenAI API 형식에 맞게 메시지 포맷팅"""
messages = []
# 시스템 메시지가 없으면 추가
has_system_message = any(msg["role"] == "system" for msg in self.conversations.get(user_id, []))
if not has_system_message:
messages.append({"role": "system", "content": "You are a helpful assistant."})
# 기존 메시지 추가
for msg in self.conversations.get(user_id, []):
if "timestamp" in msg: # timestamp 필드 제거
msg_copy = msg.copy()
del msg_copy["timestamp"]
messages.append(msg_copy)
else:
messages.append(msg)
return messages
def clear_conversation(self, user_id):
"""특정 사용자의 대화 기록 초기화"""
if user_id in self.conversations:
# 시스템 메시지만 유지
self.conversations[user_id] = [
msg for msg in self.conversations[user_id]
if msg["role"] == "system"
]
def save_conversations(self, filepath):
"""대화 기록을 파일로 저장"""
with open(filepath, 'w', encoding='utf-8') as f:
json.dump(self.conversations, f, ensure_ascii=False, indent=2)
def load_conversations(self, filepath):
"""파일에서 대화 기록 로드"""
try:
with open(filepath, 'r', encoding='utf-8') as f:
self.conversations = json.load(f)
except (FileNotFoundError, json.JSONDecodeError):
self.conversations = {}
4.3 감정 분석 통합
from transformers import pipeline
class EmotionAnalyzer:
def __init__(self):
# 감정 분석 모델 로드
self.emotion_analyzer = pipeline(
"text-classification",
model="j-hartmann/emotion-english-distilroberta-base",
top_k=None
)
def analyze_emotion(self, text):
# 영어로만 작동하므로 필요시 번역 로직 추가
try:
results = self.emotion_analyzer(text)
emotions = {item['label']: item['score'] for item in results[0]}
# 가장 높은 감정 추출
top_emotion = max(emotions.items(), key=lambda x: x[1])
return {
'top_emotion': top_emotion[0],
'top_score': top_emotion[1],
'all_emotions': emotions
}
except Exception as e:
print(f"감정 분석 오류: {e}")
return {
'top_emotion': 'neutral',
'top_score': 1.0,
'all_emotions': {'neutral': 1.0}
}
def get_emotion_based_response(self, emotion_result):
# 감정에 따른 응답 템플릿
emotion_responses = {
'joy': [
"기쁜 소식이네요!",
"정말 좋은 일이군요.",
"함께 기뻐할 수 있어 행복해요."
],
'sadness': [
"그런 일이 있으셨군요. 힘드시겠어요.",
"마음이 무거우실 것 같아요.",
"함께 이야기 나누다 보면 마음이 조금 나아질 수 있을 거예요."
],
'anger': [
"화가 나셨군요. 충분히 이해됩니다.",
"그런 상황에서 화가 나는 건 자연스러운 일이에요.",
"감정을 표현해주셔서 감사합니다."
],
'fear': [
"걱정되는 일이 있으신가요?",
"불안한 마음이 드실 수 있어요.",
"함께 걱정을 나눠서 조금이나마 도움이 되고 싶어요."
],
'surprise': [
"정말 놀라운 일이군요!",
"예상치 못한 일이 있으셨나봐요.",
"저도 함께 놀랐어요."
],
'neutral': [
"네, 말씀해주세요.",
"어떻게 도와드릴까요?",
"더 자세히 알려주시겠어요?"
]
}
# 감정에 해당하는 응답이 없으면 중립 응답 사용
responses = emotion_responses.get(emotion_result['top_emotion'], emotion_responses['neutral'])
return np.random.choice(responses)
# 감정 분석을 통합한 응답 생성 예시
def generate_empathetic_response(user_input):
# 감정 분석
emotion_analyzer = EmotionAnalyzer()
emotion_result = emotion_analyzer.analyze_emotion(user_input)
# 기본 응답 생성 (여기서는 예시로 간단히 구현)
base_response = f"입력: {user_input}"
# 감정 기반 응답 요소 생성
emotion_response = emotion_analyzer.get_emotion_based_response(emotion_result)
# 응답 조합
final_response = f"{emotion_response} {base_response}"
return final_response, emotion_result
4.4 지식 베이스 통합
import faiss
import numpy as np
from sentence_transformers import SentenceTransformer
import pandas as pd
class KnowledgeBase:
def __init__(self, embedding_model='sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2'):
# 임베딩 모델 로드
self.model = SentenceTransformer(embedding_model)
self.index = None
self.documents = []
def add_documents(self, documents, rebuild_index=True):
"""지식 베이스에 문서 추가"""
self.documents.extend(documents)
if rebuild_index:
self.build_index()
def build_index(self):
"""FAISS 인덱스 구축"""
# 문서 임베딩 생성
embeddings = self.model.encode([doc['content'] for doc in self.documents])
# FAISS 인덱스 생성
dimension = embeddings.shape[1]
self.index = faiss.IndexFlatL2(dimension)
# 정규화된 벡터로 변환
faiss.normalize_L2(embeddings)
# 인덱스에 벡터 추가
self.index.add(embeddings)
def search(self, query, top_k=3):
"""쿼리와 관련된 문서 검색"""
if not self.index or not self.documents:
return []
# 쿼리 임베딩 생성
query_embedding = self.model.encode([query])
# 정규화
faiss.normalize_L2(query_embedding)
# 유사도 검색
distances, indices = self.index.search(query_embedding, top_k)
# 결과 반환
results = []
for i, idx in enumerate(indices[0]):
if idx < len(self.documents):
doc = self.documents[idx].copy()
doc['score'] = 1 - distances[0][i] # 거리를 유사도 점수로 변환
results.append(doc)
return results
def save(self, index_path, documents_path):
"""인덱스와 문서 저장"""
# 인덱스 저장
if self.index:
faiss.write_index(self.index, index_path)
# 문서 저장
pd.DataFrame(self.documents).to_csv(documents_path, index=False)
def load(self, index_path, documents_path):
"""인덱스와 문서 로드"""
# 문서 로드
df = pd.read_csv(documents_path)
self.documents = df.to_dict('records')
# 인덱스 로드
self.index = faiss.read_index(index_path)
# 예시: 지식 베이스 구축 및 검색
def build_sample_knowledge_base():
# 샘플 FAQ 문서
sample_documents = [
{
'id': 1,
'title': '환불 정책',
'content': '구매 후 7일 이내에 제품에 문제가 없는 경우 전액 환불이 가능합니다. 제품에 문제가 있는 경우 30일 이내 교환 또는 환불이 가능합니다.',
'category': 'customer_service'
},
{
'id': 2,
'title': '배송 정책',
'content': '국내 배송은 주문 확인
'오픈소스를 위한 기초 상식' 카테고리의 다른 글
| LangChain 기초 학습 가이드 (0) | 2025.03.30 |
|---|---|
| OpenAI API 기초 학습 가이드 (0) | 2025.03.29 |
| 데이터 대시보드 제작 가이드 (0) | 2025.03.28 |
| 자동 리포트 생성 시스템 학습 자료 (0) | 2025.03.27 |
| 데이터 수집-분석-저장 파이프라인 구축 가이드 (0) | 2025.03.26 |
