Mem0 Multi-User Collaboration: AI 메모리로 구축하는 팀 협업 시스템
개요
Mem0는 AI 에이전트를 위한 메모리 레이어로, 다중 사용자 협업 시스템을 구축할 때 강력한 기능을 제공합니다. 이 튜토리얼에서는 Mem0와 OpenAI를 활용하여 팀 협업 채팅 및 작업 관리 시스템을 구축하는 방법을 단계별로 설명합니다.
주요 기능
- ✅ 메시지 귀속: 각 메시지가 작성자에게 명확히 귀속
- ✅ 공유 프로젝트 공간: 모든 팀원이 접근 가능한 공유 메모리
- ✅ 기여도 추적: 누가 무엇을 기여했는지 추적
- ✅ 실시간 협업: 메시지 정렬 및 그룹화로 효율적 협업
- ✅ AI 어시스턴트: 프로젝트 컨텍스트 기반 지능형 응답
환경 설정
필수 패키지 설치
pip install openai mem0ai python-dotenv
환경 변수 설정
# .env 파일 생성
echo "OPENAI_API_KEY=sk-your-openai-api-key" > .env
API 키 설정 확인
import os
from dotenv import load_dotenv
# 환경 변수 로드
load_dotenv()
# API 키 확인
if not os.getenv("OPENAI_API_KEY"):
print("⚠️ OPENAI_API_KEY가 설정되지 않았습니다!")
else:
print("✅ OpenAI API 키가 설정되었습니다.")
Mem0 연구 성과
최신 연구 결과에 따르면, Mem0는 OpenAI Memory 대비 다음과 같은 성능 향상을 보여줍니다:
- 📈 26% 높은 정확도
- ⚡ 91% 낮은 지연시간
- 💰 90% 토큰 절약
이러한 성능 향상으로 실시간 협업 시스템에 최적화되어 있습니다.
핵심 클래스 구현
CollaborativeAgent 클래스
from openai import OpenAI
from mem0 import Memory
import os
from datetime import datetime
from collections import defaultdict
from dotenv import load_dotenv
# 환경 변수 로드
load_dotenv()
class CollaborativeAgent:
def __init__(self, run_id):
"""
협업 에이전트 초기화
Args:
run_id (str): 프로젝트별 고유 식별자
"""
self.run_id = run_id
self.mem = Memory()
self.client = OpenAI()
print(f"🚀 협업 에이전트 시작됨 - 프로젝트 ID: {run_id}")
def add_message(self, role, name, content):
"""
메시지를 메모리에 추가
Args:
role (str): 사용자 역할 (user/assistant)
name (str): 메시지 작성자명
content (str): 메시지 내용
"""
msg = {
"role": role,
"name": name,
"content": content
}
# 메모리에 메시지 저장 (추론 비활성화로 성능 향상)
self.mem.add([msg], run_id=self.run_id, infer=False)
print(f"💬 [{name}] 메시지 추가됨: {content[:50]}...")
def brainstorm(self, prompt):
"""
AI 어시스턴트의 브레인스토밍 응답 생성
Args:
prompt (str): 사용자 질문/요청
Returns:
str: AI 어시스턴트 응답
"""
# 관련 메모리 검색 (최근 5개)
memories = self.mem.search(
prompt,
run_id=self.run_id,
limit=5
)["results"]
# 컨텍스트 구성
context = "\n".join([
f"- {m['memory']} (by {m.get('actor_id', 'Unknown')})"
for m in memories
])
# AI 응답 생성
messages = [
{
"role": "system",
"content": "당신은 팀 프로젝트를 돕는 유능한 AI 어시스턴트입니다. 한국어로 명확하고 도움이 되는 응답을 제공하세요."
},
{
"role": "user",
"content": f"질문: {prompt}\n\n프로젝트 컨텍스트:\n{context}"
}
]
response = self.client.chat.completions.create(
model="gpt-4o-mini",
messages=messages,
temperature=0.7
)
reply = response.choices[0].message.content.strip()
# AI 응답도 메모리에 저장
self.add_message("assistant", "AI_Assistant", reply)
return reply
def get_all_messages(self):
"""모든 메시지 조회"""
return self.mem.get_all(run_id=self.run_id)["results"]
def print_sorted_by_time(self):
"""시간순으로 정렬된 메시지 출력"""
messages = self.get_all_messages()
messages.sort(key=lambda m: m.get('created_at', ''))
print("\n" + "="*60)
print("📅 시간순 메시지 목록")
print("="*60)
for m in messages:
who = m.get("actor_id") or "Unknown"
ts = m.get('created_at', 'Timestamp N/A')
try:
# ISO 형식 타임스탬프 파싱
dt = datetime.fromisoformat(ts.replace('Z', '+00:00'))
ts_fmt = dt.strftime('%Y-%m-%d %H:%M:%S')
except Exception:
ts_fmt = ts
print(f"[{ts_fmt}] [{who}] {m['memory']}")
def print_grouped_by_actor(self):
"""작성자별로 그룹화된 메시지 출력"""
messages = self.get_all_messages()
grouped = defaultdict(list)
for m in messages:
actor = m.get("actor_id") or "Unknown"
grouped[actor].append(m)
print("\n" + "="*60)
print("👥 작성자별 메시지 그룹")
print("="*60)
for actor, mems in grouped.items():
print(f"\n🔸 {actor} ({len(mems)}개 메시지)")
print("-" * 40)
for m in mems:
ts = m.get('created_at', 'Timestamp N/A')
try:
dt = datetime.fromisoformat(ts.replace('Z', '+00:00'))
ts_fmt = dt.strftime('%H:%M:%S')
except Exception:
ts_fmt = ts
print(f" [{ts_fmt}] {m['memory']}")
def get_project_summary(self):
"""프로젝트 요약 생성"""
all_messages = self.get_all_messages()
if not all_messages:
return "📝 아직 메시지가 없습니다."
# 최근 메시지들을 기반으로 요약 생성
recent_content = "\n".join([
f"- {m['memory']}" for m in all_messages[-10:]
])
summary_prompt = f"다음 프로젝트 대화를 간단히 요약해주세요:\n{recent_content}"
summary = self.brainstorm(summary_prompt)
return summary
def search_messages(self, query, limit=5):
"""특정 키워드로 메시지 검색"""
results = self.mem.search(
query,
run_id=self.run_id,
limit=limit
)["results"]
print(f"\n🔍 '{query}' 검색 결과 ({len(results)}개)")
print("-" * 40)
for result in results:
actor = result.get("actor_id", "Unknown")
memory = result['memory']
print(f"[{actor}] {memory}")
return results
실제 사용 예제
기본 사용법
# 프로젝트 ID 설정 (팀별로 고유하게 설정)
PROJECT_ID = "landing-page-project-2025"
# 협업 에이전트 초기화
agent = CollaborativeAgent(PROJECT_ID)
# 팀원들의 메시지 추가
agent.add_message("user", "Alice", "새 랜딩 페이지 작업 목록을 정리해봅시다.")
agent.add_message("user", "Bob", "히어로 섹션 카피라이팅을 제가 담당하겠습니다.")
agent.add_message("user", "Carol", "제품 스크린샷 선별은 제가 할게요.")
agent.add_message("user", "David", "반응형 CSS 작업을 맡겠습니다.")
print("✅ 초기 메시지들이 추가되었습니다.")
AI 어시스턴트와 브레인스토밍
# AI 어시스턴트에게 질문
questions = [
"현재 미완료 작업들이 무엇인가요?",
"각 팀원의 담당 업무를 정리해주세요.",
"랜딩 페이지 제작에 추가로 필요한 작업은 무엇인가요?"
]
for question in questions:
print(f"\n❓ 질문: {question}")
print("🤖 AI 응답:")
response = agent.brainstorm(question)
print(response)
print("-" * 50)
고급 협업 기능
def demonstrate_advanced_features():
"""고급 협업 기능 시연"""
# 추가 팀 활동 시뮬레이션
additional_messages = [
("user", "Alice", "디자인 시스템 색상 팔레트를 확정했습니다. #3B82F6, #EF4444, #10B981을 사용합니다."),
("user", "Bob", "히어로 섹션 초안 완료했습니다. 검토 부탁드립니다."),
("user", "Carol", "스크린샷 10개 선별 완료. Figma에 업로드했습니다."),
("user", "David", "모바일 반응형 작업 80% 완료. 태블릿 뷰 작업 중입니다."),
("user", "Alice", "QA 테스팅은 언제 시작할까요?")
]
for role, name, content in additional_messages:
agent.add_message(role, name, content)
print("📝 추가 팀 활동이 기록되었습니다.\n")
# 1. 시간순 메시지 조회
agent.print_sorted_by_time()
# 2. 작성자별 메시지 그룹화
agent.print_grouped_by_actor()
# 3. 특정 키워드 검색
search_terms = ["히어로", "디자인", "반응형"]
for term in search_terms:
agent.search_messages(term, limit=3)
# 4. 프로젝트 요약
print("\n📊 프로젝트 요약")
print("="*40)
summary = agent.get_project_summary()
print(summary)
# 고급 기능 실행
demonstrate_advanced_features()
실용적 활용 사례
1. 작업 상태 추적 시스템
class TaskTracker(CollaborativeAgent):
def __init__(self, run_id):
super().__init__(run_id)
self.task_statuses = ["TODO", "IN_PROGRESS", "REVIEW", "DONE"]
def add_task(self, assignee, task_title, description="", status="TODO"):
"""새 작업 추가"""
task_msg = f"[작업] {task_title} - 담당자: {assignee}, 상태: {status}"
if description:
task_msg += f", 설명: {description}"
self.add_message("user", assignee, task_msg)
print(f"✅ 작업 추가됨: {task_title}")
def update_task_status(self, assignee, task_title, new_status):
"""작업 상태 업데이트"""
if new_status not in self.task_statuses:
print(f"❌ 잘못된 상태: {new_status}")
return
update_msg = f"[업데이트] {task_title} 상태 변경: {new_status}"
self.add_message("user", assignee, update_msg)
print(f"🔄 상태 업데이트: {task_title} -> {new_status}")
def get_tasks_by_status(self, status):
"""상태별 작업 조회"""
results = self.search_messages(f"상태: {status}", limit=10)
return [r for r in results if status in r['memory']]
# 작업 추적 시스템 사용 예제
task_tracker = TaskTracker("web-project-tasks")
# 작업 추가
task_tracker.add_task("Alice", "랜딩 페이지 와이어프레임", "메인 페이지 레이아웃 설계")
task_tracker.add_task("Bob", "로고 디자인", "브랜드 아이덴티티에 맞는 로고")
task_tracker.add_task("Carol", "콘텐츠 작성", "서비스 소개 텍스트")
# 상태 업데이트
task_tracker.update_task_status("Alice", "랜딩 페이지 와이어프레임", "IN_PROGRESS")
task_tracker.update_task_status("Bob", "로고 디자인", "DONE")
2. 회의록 관리 시스템
class MeetingManager(CollaborativeAgent):
def __init__(self, run_id):
super().__init__(run_id)
def start_meeting(self, meeting_title, attendees):
"""회의 시작"""
meeting_msg = f"[회의 시작] {meeting_title} - 참석자: {', '.join(attendees)}"
self.add_message("system", "MeetingBot", meeting_msg)
print(f"🗓️ 회의 시작: {meeting_title}")
def add_agenda_item(self, presenter, topic, details=""):
"""안건 추가"""
agenda_msg = f"[안건] {topic} - 발표자: {presenter}"
if details:
agenda_msg += f", 내용: {details}"
self.add_message("user", presenter, agenda_msg)
def add_action_item(self, assignee, action, deadline=""):
"""액션 아이템 추가"""
action_msg = f"[액션] {action} - 담당: {assignee}"
if deadline:
action_msg += f", 마감: {deadline}"
self.add_message("user", assignee, action_msg)
def end_meeting(self):
"""회의 종료 및 요약"""
summary = self.brainstorm("이번 회의의 주요 내용과 액션 아이템을 요약해주세요.")
self.add_message("system", "MeetingBot", f"[회의 요약] {summary}")
return summary
# 회의 관리 시스템 사용 예제
meeting = MeetingManager("weekly-standup-2025-06")
meeting.start_meeting("주간 스탠드업", ["Alice", "Bob", "Carol", "David"])
meeting.add_agenda_item("Alice", "지난주 진행 상황", "완료된 작업 및 이슈 공유")
meeting.add_agenda_item("Bob", "이번주 계획", "우선순위 작업 논의")
meeting.add_action_item("Carol", "클라이언트 피드백 수집", "2025-06-20")
meeting.add_action_item("David", "서버 성능 최적화", "2025-06-25")
summary = meeting.end_meeting()
print(f"\n📋 회의 요약:\n{summary}")
성능 최적화 팁
1. 메모리 효율성
# 대량 메시지 처리 시 배치 추가
def add_messages_batch(agent, messages_list):
"""메시지 배치 추가로 성능 향상"""
batch_msgs = []
for role, name, content in messages_list:
batch_msgs.append({
"role": role,
"name": name,
"content": content
})
# 한 번에 배치 처리
agent.mem.add(batch_msgs, run_id=agent.run_id, infer=False)
print(f"✅ {len(batch_msgs)}개 메시지 배치 추가 완료")
# 사용 예제
batch_messages = [
("user", "Alice", "첫 번째 메시지"),
("user", "Bob", "두 번째 메시지"),
("user", "Carol", "세 번째 메시지")
]
add_messages_batch(agent, batch_messages)
2. 검색 성능 최적화
def optimized_search(agent, query, limit=5, filter_by_actor=None):
"""최적화된 검색 함수"""
results = agent.mem.search(
query,
run_id=agent.run_id,
limit=limit
)["results"]
# 특정 작성자로 필터링 (옵션)
if filter_by_actor:
results = [r for r in results if r.get("actor_id") == filter_by_actor]
return results
# 사용 예제
alice_results = optimized_search(agent, "디자인", limit=3, filter_by_actor="Alice")
print(f"Alice의 디자인 관련 메시지: {len(alice_results)}개")
보안 및 권한 관리
사용자 권한 시스템
class SecureCollaborativeAgent(CollaborativeAgent):
def __init__(self, run_id, admin_users=None):
super().__init__(run_id)
self.admin_users = admin_users or []
self.user_permissions = {}
def set_user_permission(self, username, permission_level):
"""사용자 권한 설정 (admin, member, viewer)"""
self.user_permissions[username] = permission_level
print(f"🔐 {username} 권한 설정: {permission_level}")
def add_message_with_permission(self, role, name, content):
"""권한 확인 후 메시지 추가"""
user_permission = self.user_permissions.get(name, "viewer")
if user_permission == "viewer":
print(f"❌ {name}은 읽기 전용 권한입니다.")
return False
self.add_message(role, name, content)
return True
def delete_message_if_admin(self, admin_name, message_id):
"""관리자만 메시지 삭제 가능"""
if admin_name not in self.admin_users:
print(f"❌ {admin_name}은 관리자가 아닙니다.")
return False
# 실제 삭제 로직 (Mem0 API에 따라 구현)
print(f"🗑️ 관리자 {admin_name}이 메시지를 삭제했습니다.")
return True
# 보안 시스템 사용 예제
secure_agent = SecureCollaborativeAgent(
"secure-project",
admin_users=["Alice", "ProjectManager"]
)
secure_agent.set_user_permission("Alice", "admin")
secure_agent.set_user_permission("Bob", "member")
secure_agent.set_user_permission("Viewer1", "viewer")
# 권한에 따른 메시지 추가 테스트
secure_agent.add_message_with_permission("user", "Alice", "관리자 메시지") # 성공
secure_agent.add_message_with_permission("user", "Bob", "멤버 메시지") # 성공
secure_agent.add_message_with_permission("user", "Viewer1", "뷰어 메시지") # 실패
트러블슈팅
일반적인 문제 해결
def diagnose_system(agent):
"""시스템 진단 및 문제 해결"""
print("🔍 시스템 진단 시작...")
try:
# 1. 메모리 연결 확인
all_messages = agent.get_all_messages()
print(f"✅ 메모리 연결 정상 - {len(all_messages)}개 메시지")
# 2. OpenAI API 연결 확인
test_response = agent.brainstorm("테스트 메시지입니다.")
print("✅ OpenAI API 연결 정상")
# 3. 검색 기능 확인
search_results = agent.search_messages("테스트", limit=1)
print("✅ 검색 기능 정상")
print("🎉 모든 시스템이 정상 작동중입니다!")
except Exception as e:
print(f"❌ 오류 발생: {str(e)}")
print("해결 방법:")
print("1. API 키 확인")
print("2. 인터넷 연결 확인")
print("3. 패키지 버전 확인")
# 진단 실행
diagnose_system(agent)
데이터 백업 및 복원
import json
from datetime import datetime
def backup_project_data(agent, backup_filename=None):
"""프로젝트 데이터 백업"""
if not backup_filename:
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
backup_filename = f"backup_{agent.run_id}_{timestamp}.json"
all_messages = agent.get_all_messages()
backup_data = {
"project_id": agent.run_id,
"backup_date": datetime.now().isoformat(),
"message_count": len(all_messages),
"messages": all_messages
}
with open(backup_filename, 'w', encoding='utf-8') as f:
json.dump(backup_data, f, ensure_ascii=False, indent=2)
print(f"💾 백업 완료: {backup_filename}")
return backup_filename
def restore_project_data(agent, backup_filename):
"""백업 데이터 복원"""
try:
with open(backup_filename, 'r', encoding='utf-8') as f:
backup_data = json.load(f)
messages = backup_data['messages']
# 메시지 복원 (실제 구현은 Mem0 API에 따라 달라질 수 있음)
print(f"🔄 {len(messages)}개 메시지 복원 중...")
for msg in messages:
# 원본 메시지 형식으로 복원
content = msg['memory']
actor = msg.get('actor_id', 'Unknown')
agent.add_message("user", actor, content)
print(f"✅ 복원 완료: {backup_filename}")
except Exception as e:
print(f"❌ 복원 실패: {str(e)}")
# 백업 및 복원 예제
backup_file = backup_project_data(agent)
# restore_project_data(agent, backup_file) # 필요시 복원
최적화된 워크플로우
전체 협업 워크플로우 예제
def complete_collaboration_workflow():
"""완전한 협업 워크플로우 데모"""
# 1. 프로젝트 초기화
project = CollaborativeAgent("complete-workflow-demo")
# 2. 팀 설정
team_members = ["Alice", "Bob", "Carol", "David"]
project.add_message("system", "ProjectBot",
f"프로젝트 시작 - 팀원: {', '.join(team_members)}")
# 3. 초기 브레인스토밍
project.add_message("user", "Alice", "새 모바일 앱 프로젝트 킥오프 미팅입니다.")
project.add_message("user", "Bob", "UI/UX 디자인 담당하겠습니다.")
project.add_message("user", "Carol", "백엔드 API 개발 맡겠습니다.")
project.add_message("user", "David", "프론트엔드 개발과 배포 담당하겠습니다.")
# 4. AI 어시스턴트 질문
planning_response = project.brainstorm(
"모바일 앱 개발 프로젝트의 주요 마일스톤과 일정을 제안해주세요."
)
print(f"📋 AI 계획 제안:\n{planning_response}\n")
# 5. 진행 상황 업데이트
progress_updates = [
("user", "Bob", "와이어프레임 초안 완성. Figma 링크 공유했습니다."),
("user", "Carol", "데이터베이스 스키마 설계 완료. 리뷰 요청드립니다."),
("user", "David", "React Native 환경 설정 완료."),
("user", "Alice", "클라이언트 요구사항 추가 전달받았습니다.")
]
for role, name, content in progress_updates:
project.add_message(role, name, content)
# 6. 종합 분석
print("📊 프로젝트 현황 분석")
print("="*50)
# 시간순 메시지
project.print_sorted_by_time()
# 작성자별 기여도
project.print_grouped_by_actor()
# 키워드 검색
project.search_messages("완료", limit=5)
# 최종 요약
final_summary = project.get_project_summary()
print(f"\n📝 프로젝트 최종 요약:\n{final_summary}")
return project
# 전체 워크플로우 실행
complete_project = complete_collaboration_workflow()
결론
Mem0 Multi-User Collaboration 시스템은 팀 협업의 효율성을 크게 향상시킬 수 있는 강력한 도구입니다.
주요 장점
- 🚀 높은 성능: OpenAI Memory 대비 26% 높은 정확도, 91% 낮은 지연시간
- 💡 유연한 구조: 다양한 협업 시나리오에 맞게 확장 가능
- 🔒 보안: 사용자 권한 관리 및 데이터 보호
- 📊 분석: 팀 기여도 추적 및 프로젝트 인사이트
실제 적용 분야
- 소프트웨어 개발 팀: 코드 리뷰, 버그 추적, 스프린트 계획
- 디자인 팀: 크리에이티브 브레인스토밍, 피드백 수집
- 마케팅 팀: 캠페인 기획, 콘텐츠 협업
- 교육 기관: 그룹 프로젝트, 학습 진도 관리
다음 단계
이 튜토리얼을 통해 기본적인 협업 시스템을 구축했다면, 다음과 같은 고급 기능들을 추가로 구현해볼 수 있습니다:
- 📱 실시간 웹 인터페이스 (Streamlit, FastAPI)
- 🔔 알림 시스템 (이메일, Slack 연동)
- 📈 대시보드 (프로젝트 진행률, 팀 성과 지표)
- 🤖 고급 AI 기능 (자동 작업 할당, 위험 요소 감지)
Mem0의 강력한 메모리 기능을 활용하여 더욱 스마트한 협업 환경을 구축해보세요!