1. 기본 설정
먼저 필요한 라이브러리와 API 키를 설정해야 합니다.
import openai
from datetime import datetime
import json
import os
# API 키 설정
openai.api_key = 'your-api-key' # 실제 사용시에는 환경 변수로 관리하는 것을 추천합니다
2. 기본 대화 기능 구현
class PersonalAssistant:
def __init__(self):
self.conversation_history = []
self.model = "gpt-3.5-turbo" # 또는 "gpt-4"
def add_to_history(self, role, content):
self.conversation_history.append({
"role": role,
"content": content
})
def get_response(self, user_input):
# 사용자 입력 추가
self.add_to_history("user", user_input)
try:
# API 호출
response = openai.ChatCompletion.create(
model=self.model,
messages=self.conversation_history
)
# 응답 추출 및 히스토리 추가
assistant_response = response.choices[0].message.content
self.add_to_history("assistant", assistant_response)
return assistant_response
except Exception as e:
return f"Error: {str(e)}"
3. 작업 관리 기능 추가
class TaskManager:
def __init__(self):
self.tasks = []
def add_task(self, task, deadline=None):
self.tasks.append({
"task": task,
"deadline": deadline,
"created_at": datetime.now().isoformat(),
"completed": False
})
def mark_completed(self, task_index):
if 0 <= task_index < len(self.tasks):
self.tasks[task_index]["completed"] = True
def get_active_tasks(self):
return [task for task in self.tasks if not task["completed"]]
class EnhancedPersonalAssistant(PersonalAssistant):
def __init__(self):
super().__init__()
self.task_manager = TaskManager()
def process_command(self, command):
# 명령어 처리 로직
if command.startswith("할일추가:"):
task = command[5:].strip()
self.task_manager.add_task(task)
return f"할일이 추가되었습니다: {task}"
elif command == "할일목록":
tasks = self.task_manager.get_active_tasks()
return "\n".join([f"{i+1}. {task['task']}" for i, task in enumerate(tasks)])
else:
return self.get_response(command)
4. 일정 관리 기능 통합
class Calendar:
def __init__(self):
self.events = []
def add_event(self, title, date, time=None):
self.events.append({
"title": title,
"date": date,
"time": time,
"created_at": datetime.now().isoformat()
})
def get_events_for_date(self, date):
return [event for event in self.events if event["date"] == date]
class CompletePersonalAssistant(EnhancedPersonalAssistant):
def __init__(self):
super().__init__()
self.calendar = Calendar()
def process_command(self, command):
if command.startswith("일정추가:"):
# 일정 추가 명령 처리
try:
parts = command[5:].strip().split(',')
title = parts[0].strip()
date = parts[1].strip()
time = parts[2].strip() if len(parts) > 2 else None
self.calendar.add_event(title, date, time)
return f"일정이 추가되었습니다: {title} ({date} {time if time else ''})"
except Exception as e:
return f"일정 추가 중 오류가 발생했습니다: {str(e)}"
elif command.startswith("일정확인:"):
# 특정 날짜의 일정 확인
date = command[5:].strip()
events = self.calendar.get_events_for_date(date)
if events:
return "\n".join([f"- {event['title']} ({event['time'] if event['time'] else '시간 미정'})"
for event in events])
else:
return f"{date}에 예정된 일정이 없습니다."
else:
return super().process_command(command)
5. 사용 예시
def main():
assistant = CompletePersonalAssistant()
while True:
user_input = input("무엇을 도와드릴까요? (종료하려면 'quit' 입력): ")
if user_input.lower() == 'quit':
break
response = assistant.process_command(user_input)
print("\n비서:", response, "\n")
if __name__ == "__main__":
main()
6. 데이터 영속성 추가
class DataManager:
def __init__(self, filename="assistant_data.json"):
self.filename = filename
def save_data(self, data):
with open(self.filename, 'w', encoding='utf-8') as f:
json.dump(data, f, ensure_ascii=False, indent=2)
def load_data(self):
try:
with open(self.filename, 'r', encoding='utf-8') as f:
return json.load(f)
except FileNotFoundError:
return {"tasks": [], "events": [], "conversation_history": []}
class PersistentPersonalAssistant(CompletePersonalAssistant):
def __init__(self):
super().__init__()
self.data_manager = DataManager()
self.load_state()
def load_state(self):
data = self.data_manager.load_data()
self.task_manager.tasks = data["tasks"]
self.calendar.events = data["events"]
self.conversation_history = data["conversation_history"]
def save_state(self):
data = {
"tasks": self.task_manager.tasks,
"events": self.calendar.events,
"conversation_history": self.conversation_history
}
self.data_manager.save_data(data)
def process_command(self, command):
response = super().process_command(command)
self.save_state()
return response
활용 팁
- 환경 변수 설정
# .env 파일 생성
OPENAI_API_KEY=your-api-key
# Python에서 로드
from dotenv import load_dotenv
load_dotenv()
openai.api_key = os.getenv('OPENAI_API_KEY')
- 에러 처리 강화
def safe_api_call(func):
def wrapper(*args, **kwargs):
try:
return func(*args, **kwargs)
except openai.error.RateLimitError:
return "API 호출 한도를 초과했습니다. 잠시 후 다시 시도해주세요."
except openai.error.InvalidRequestError as e:
return f"잘못된 요청입니다: {str(e)}"
except Exception as e:
return f"예기치 않은 오류가 발생했습니다: {str(e)}"
return wrapper
주의사항
- API 키 보안
- 절대로 코드에 직접 API 키를 하드코딩하지 마세요
- 환경 변수나 보안 저장소를 사용하세요
- 요청 제한 관리
- API 호출 횟수를 모니터링하세요
- 적절한 캐싱 전략을 구현하세요
- 데이터 백업
- 정기적으로 데이터를 백업하세요
- 중요한 정보는 암호화하여 저장하세요
확장 가능성
- 음성 인식 통합
- 웹 인터페이스 추가
- 다중 사용자 지원
- 자동화된 태스크 실행
- 외부 서비스 연동
이 코드는 기본적인 개인 비서 기능을 구현한 것이며, 필요에 따라 기능을 추가하거나 수정할 수 있습니다. 실제 사용 시에는 보안과 에러 처리를 더욱 강화하시기 바랍니다.