Cognee 완벽 가이드 - AI 에이전트를 위한 메모리 시스템 구축하기

개요

AI 에이전트가 과거 대화와 학습 내용을 기억하고 활용할 수 있다면 어떨까요? Cognee는 바로 이런 꿈을 현실로 만들어주는 혁신적인 오픈소스 프로젝트입니다. 단 5줄의 코드로 AI 에이전트에 메모리를 부여하고, 기존의 RAG 시스템을 뛰어넘는 성능을 제공합니다.

Cognee란 무엇인가?

Cognee는 AI 에이전트를 위한 동적 메모리 시스템으로, ECL(Extract, Cognify, Load) 파이프라인을 통해 확장 가능하고 모듈형 구조를 제공합니다.

핵심 특징

• 🧠 지능형 메모리: 과거 대화, 문서, 이미지, 오디오 전사본을 상호 연결
• 🔄 RAG 대체: 기존 RAG 시스템을 대체하여 개발 노력과 비용 절감
• 📊 다중 데이터베이스: 그래프 및 벡터 데이터베이스에 Pydantic만으로 데이터 로드
• 🔌 다양한 데이터 소스: 30개 이상의 데이터 소스에서 데이터 수집 및 조작

설치하기

요구사항

• Python 3.8 ~ 3.12
• OpenAI API 키 (또는 다른 LLM 제공자)

pip로 설치

pip install cognee

UV로 모든 옵션 포함 설치

# 레포지토리 클론
git clone https://github.com/topoteretes/cognee.git
cd cognee

# UV로 모든 종속성 설치
uv sync --all-extras

Poetry로 설치

git clone https://github.com/topoteretes/cognee.git
cd cognee
poetry install --all-extras

기본 설정

환경변수 설정

import os

# OpenAI API 키 설정
os.environ["LLM_API_KEY"] = "your-openai-api-key"

# 또는 .env 파일 사용
# LLM_API_KEY=your-openai-api-key

다른 LLM 제공자 설정

# Anthropic Claude 사용
os.environ["LLM_API_KEY"] = "your-anthropic-api-key"
os.environ["LLM_PROVIDER"] = "anthropic"

# Local Ollama 사용
os.environ["LLM_PROVIDER"] = "ollama"
os.environ["LLM_ENDPOINT"] = "http://localhost:11434"

첫 번째 예제: 간단한 지식 그래프 생성

import cognee
import asyncio

async def basic_example():
    """기본적인 Cognee 사용 예제"""
    
    # 1. 텍스트 데이터 추가
    text = """
    자연어 처리(NLP)는 컴퓨터 과학과 정보 검색의 학제간 하위 분야입니다.
    NLP는 컴퓨터와 인간 언어 간의 상호작용에 중점을 두며,
    기계가 자연어를 이해하고 처리할 수 있게 합니다.
    """
    
    await cognee.add(text)
    
    # 2. 지식 그래프 생성 (cognify)
    await cognee.cognify()
    
    # 3. 쿼리 실행
    results = await cognee.search("NLP에 대해 설명해주세요")
    
    # 4. 결과 출력
    for result in results:
        print(f"결과: {result}")

# 실행
if __name__ == "__main__":
    asyncio.run(basic_example())

고급 사용법

다양한 데이터 소스 활용

import cognee
import asyncio

async def advanced_data_sources():
    """다양한 데이터 소스를 활용한 예제"""
    
    # 파일 업로드
    await cognee.add("/path/to/document.pdf")
    await cognee.add("/path/to/image.jpg")
    
    # URL에서 데이터 수집
    await cognee.add("https://example.com/article")
    
    # 여러 텍스트 데이터 배치 추가
    documents = [
        "머신러닝은 인공지능의 한 분야입니다.",
        "딥러닝은 머신러닝의 하위 분야입니다.",
        "신경망은 딥러닝의 핵심 구조입니다."
    ]
    
    for doc in documents:
        await cognee.add(doc)
    
    # 지식 그래프 업데이트
    await cognee.cognify()
    
    # 복합 쿼리 실행
    results = await cognee.search("머신러닝과 딥러닝의 관계는?")
    
    return results

asyncio.run(advanced_data_sources())

사용자 정의 파이프라인 구성

import cognee
from cognee.infrastructure.databases.graph import get_graph_engine
from cognee.infrastructure.databases.vector import get_vector_engine

async def custom_pipeline():
    """사용자 정의 파이프라인 예제"""
    
    # 그래프 데이터베이스 설정
    graph_engine = await get_graph_engine()
    
    # 벡터 데이터베이스 설정  
    vector_engine = await get_vector_engine()
    
    # 데이터 추가 및 처리
    await cognee.add("Python은 프로그래밍 언어입니다.")
    
    # 커스텀 cognify 옵션
    await cognee.cognify(
        # 청크 크기 조정
        chunk_size=512,
        # 오버랩 설정
        chunk_overlap=50,
        # 임베딩 모델 지정
        embedding_model="text-embedding-ada-002"
    )
    
    # 고급 검색 옵션
    results = await cognee.search(
        query="Python 프로그래밍",
        limit=10,
        similarity_threshold=0.8
    )
    
    return results

실제 프로젝트 활용 사례

1. 개인 지식 베이스 구축

import cognee
import asyncio
import os

class PersonalKnowledgeBase:
    def __init__(self):
        self.initialized = False
    
    async def initialize(self):
        """지식 베이스 초기화"""
        if not self.initialized:
            await cognee.prune()  # 기존 데이터 정리
            self.initialized = True
    
    async def add_documents(self, file_paths):
        """문서들을 지식 베이스에 추가"""
        await self.initialize()
        
        for file_path in file_paths:
            if os.path.exists(file_path):
                await cognee.add(file_path)
                print(f"추가됨: {file_path}")
        
        await cognee.cognify()
        print("지식 그래프 생성 완료!")
    
    async def query(self, question):
        """질문에 대한 답변 검색"""
        results = await cognee.search(question)
        return results
    
    async def add_conversation(self, conversation_text):
        """대화 내용을 지식 베이스에 추가"""
        await cognee.add(f"대화 기록: {conversation_text}")
        await cognee.cognify()

# 사용 예제
async def main():
    kb = PersonalKnowledgeBase()
    
    # 문서 추가
    documents = [
        "/path/to/research_papers/",
        "/path/to/meeting_notes/",
        "/path/to/personal_docs/"
    ]
    
    await kb.add_documents(documents)
    
    # 질문하기
    answer = await kb.query("최근 연구에서 발견한 주요 인사이트는?")
    print(f"답변: {answer}")

asyncio.run(main())

2. 챗봇 메모리 시스템

import cognee
import asyncio
from datetime import datetime

class MemoryEnabledChatbot:
    def __init__(self, user_id):
        self.user_id = user_id
        self.conversation_history = []
    
    async def chat(self, user_message):
        """메모리를 활용한 대화"""
        # 사용자 메시지를 메모리에 추가
        timestamp = datetime.now().isoformat()
        conversation_entry = f"[{timestamp}] 사용자({self.user_id}): {user_message}"
        
        await cognee.add(conversation_entry)
        
        # 관련 과거 대화 검색
        context = await cognee.search(f"사용자 {self.user_id}와의 대화 중 {user_message}와 관련된 내용")
        
        # 컨텍스트를 활용한 응답 생성
        if context:
            context_text = "\n".join([str(c) for c in context[:3]])
            enhanced_prompt = f"""
            과거 대화 컨텍스트:
            {context_text}
            
            현재 사용자 질문: {user_message}
            
            위 컨텍스트를 고려하여 답변해주세요.
            """
        else:
            enhanced_prompt = user_message
        
        # 응답 생성 (여기서는 예시)
        response = f"'{user_message}'에 대한 답변입니다. (컨텍스트 활용됨)"
        
        # 봇 응답도 메모리에 저장
        bot_entry = f"[{timestamp}] 봇: {response}"
        await cognee.add(bot_entry)
        await cognee.cognify()
        
        return response

# 사용 예제
async def chatbot_demo():
    bot = MemoryEnabledChatbot("user123")
    
    # 대화 시뮬레이션
    conversations = [
        "안녕하세요! 저는 머신러닝을 공부하고 있어요.",
        "딥러닝과 머신러닝의 차이점을 알고 싶어요.",
        "이전에 말한 머신러닝 공부는 어떻게 진행되고 있나요?"
    ]
    
    for message in conversations:
        response = await bot.chat(message)
        print(f"사용자: {message}")
        print(f"봇: {response}\n")

asyncio.run(chatbot_demo())

Cognee UI 사용하기

Cognee는 웹 기반 UI도 제공합니다:

# 로컬에서 UI 실행
python cognee-gui.py

브라우저에서 http://localhost:8000으로 접속하여 시각적으로 지식 그래프를 관리할 수 있습니다.

성능 최적화 팁

1. 청킹 전략 최적화

# 문서 유형에 따른 청킹 설정
await cognee.cognify(
    chunk_size=1000,      # 긴 문서용
    chunk_overlap=100     # 컨텍스트 보존
)

# 짧은 텍스트용
await cognee.cognify(
    chunk_size=256,       # 짧은 텍스트용
    chunk_overlap=25
)

2. 메모리 관리

# 주기적으로 불필요한 데이터 정리
await cognee.prune()

# 특정 데이터만 삭제
await cognee.delete("특정 텍스트 또는 ID")

3. 배치 처리

# 대용량 데이터 배치 처리
async def batch_process(documents):
    batch_size = 10
    
    for i in range(0, len(documents), batch_size):
        batch = documents[i:i+batch_size]
        
        for doc in batch:
            await cognee.add(doc)
        
        await cognee.cognify()
        print(f"배치 {i//batch_size + 1} 처리 완료")

문제 해결

일반적인 오류와 해결방법

1. API 키 오류

# 환경변수 확인
import os
print(os.environ.get("LLM_API_KEY"))

2. 메모리 부족

# 청크 크기 줄이기
await cognee.cognify(chunk_size=256)

3. 연결 오류

# 데이터베이스 연결 확인
from cognee.infrastructure.databases import get_relational_engine

engine = await get_relational_engine()
print(f"데이터베이스 연결 상태: {engine}")

커뮤니티 및 자료

• GitHub: https://github.com/topoteretes/cognee
• Discord: 개발자 커뮤니티 참여
• Reddit: r/AIMemory 서브레딧
• 문서: 공식 문서 사이트 참조

결론

Cognee는 AI 에이전트에 메모리 기능을 부여하는 강력하면서도 사용하기 쉬운 도구입니다. 기존의 RAG 시스템을 뛰어넘는 성능과 유연성을 제공하며, 다양한 실제 프로젝트에 즉시 적용할 수 있습니다.

5.7k개의 GitHub 스타와 활발한 커뮤니티를 자랑하는 Cognee로 여러분의 AI 프로젝트에 진정한 메모리를 부여해보세요. 단 몇 줄의 코드로 시작할 수 있는 혁신적인 AI 메모리 시스템을 경험해보시기 바랍니다.

다음 단계

1. 실습: 제공된 예제 코드를 직접 실행해보세요
2. 커스터마이징: 본인의 프로젝트에 맞게 파이프라인을 수정해보세요
3. 기여: 오픈소스 프로젝트에 기여하고 커뮤니티와 함께 발전시켜보세요
4. 확장: 더 복잡한 AI 에이전트 시스템을 구축해보세요

이제 여러분도 AI 에이전트의 메모리 전문가가 되어보세요!