CrewAI 메모리 시스템

CrewAI 메모리 시스템 소개

CrewAI 프레임워크는 AI 에이전트의 기능을 크게 향상시키기 위해 정교한 메모리 시스템을 도입했습니다. 이 시스템은 단기 메모리, 장기 메모리, 엔티티 메모리, 컨텍스트 메모리로 구성되어 있으며, 각 구성 요소는 에이전트가 과거 상호작용을 기억하고, 추론하며, 학습하는 데 고유한 역할을 수행합니다.

메모리 시스템 구성 요소

구성 요소	설명
단기 메모리	RAG를 사용하여 최근 상호작용 및 결과를 일시적으로 저장하고, 현재 실행 중인 컨텍스트와 관련된 정보를 기억하여 활용할 수 있도록 합니다.
장기 메모리	이전 실행에서 얻은 귀중한 인사이트와 학습 내용을 저장하여 에이전트가 시간에 따라 지식을 축적하고 개선할 수 있도록 합니다.
엔티티 메모리	작업 중에 만난 사람, 장소, 개념 등의 정보를 캡처하고 조직화하여 더 깊은 이해와 관계 맵핑을 가능하게 합니다. 엔티티 정보 저장에 RAG를 사용합니다.
컨텍스트 메모리	단기, 장기, 엔티티 메모리를 결합하여 상호작용의 컨텍스트를 유지하며, 일련의 작업이나 대화 중에 에이전트의 응답 일관성과 적절성을 높입니다.
외부 메모리	Mem0과 같은 외부 메모리 시스템 및 공급자와 통합할 수 있어, 다양한 애플리케이션에서 전문적인 메모리 저장/검색 기능을 제공합니다. 커스텀 저장소도 지원합니다.
사용자 메모리	⚠️ 폐기 예정: 이 구성 요소는 더 이상 사용되지 않으며 향후 버전에서 제거될 예정입니다. 외부 메모리를 대신 사용하세요.

메모리 시스템이 에이전트를 강화하는 방식

1. 컨텍스트 인식: 단기 및 컨텍스트 메모리를 통해 대화나 작업 흐름 중에 컨텍스트를 유지하여 더 일관되고 관련성 있는 응답을 제공합니다.

2. 경험 축적: 장기 메모리를 통해 과거 경험으로부터 학습하고 이를 기반으로 더 나은 의사결정과 문제 해결이 가능합니다.

3. 엔티티 이해: 엔티티 메모리를 통해 핵심 개체를 인식하고 기억하여 복잡한 정보에 대해 더 정교한 처리와 상호작용이 가능합니다.

Crew에 메모리 구성 적용하기

Crew를 구성할 때 각 메모리 구성 요소를 작업 목표에 맞게 활성화하고 커스터마이징할 수 있습니다. 기본적으로 메모리 시스템은 비활성화되어 있으며, memory=True로 설정해야 활성화됩니다.

단기 메모리는 기본적으로 OpenAI 임베딩을 사용하며, embedder를 설정하여 다른 모델로 변경할 수 있습니다. 또한 직접 초기화한 메모리 인스턴스를 사용할 수도 있습니다.

• embedder 설정은 단기 메모리에만 적용되며, Chroma를 RAG에 사용합니다.
• 장기 메모리는 SQLite3를 사용하여 작업 결과를 저장하며, 현재로서는 저장 방식 변경은 불가능합니다.
• 저장 파일은 appdirs 패키지를 통해 플랫폼별 경로에 저장되며, 환경 변수 CREWAI_STORAGE_DIR를 사용해 프로젝트 이름을 변경할 수 있습니다.

예시: Crew에 메모리 설정하기

from crewai import Crew, Agent, Task, Process

# 메모리 기능이 포함된 Crew 구성
my_crew = Crew(
    agents=[...],
    tasks=[...],
    process=Process.sequential,
    memory=True,
    verbose=True
)

예시: 사용자 정의 메모리 인스턴스 사용 (예: FAISS를 VectorDB로 활용)

from crewai import Crew, Process
from crewai.memory import LongTermMemory, ShortTermMemory, EntityMemory
from crewai.memory.storage.rag_storage import RAGStorage
from crewai.memory.storage.ltm_sqlite_storage import LTMSQLiteStorage
from typing import List, Optional

# 메모리 기능이 포함된 Crew 구성
my_crew: Crew = Crew(
    agents = [...],
    tasks = [...],
    process = Process.sequential,
    memory = True,
    # 장기 메모리 설정 (세션 간 저장)
    long_term_memory = LongTermMemory(
        storage=LTMSQLiteStorage(
            db_path="/my_crew1/long_term_memory_storage.db"
        )
    ),
    # 단기 메모리 설정 (RAG 기반 현재 컨텍스트)
    short_term_memory = ShortTermMemory(
        storage = RAGStorage(
                embedder_config={
                    "provider": "openai",
                    "config": {
                        "model": 'text-embedding-3-small'
                    }
                },
                type="short_term",
                path="/my_crew1/"
            )
        ),
    # 엔티티 메모리 설정 (핵심 정보 추적)
    entity_memory = EntityMemory(
        storage=RAGStorage(
            embedder_config={
                "provider": "openai",
                "config": {
                    "model": 'text-embedding-3-small'
                }
            },
            type="short_term",
            path="/my_crew1/"
        )
    ),
    verbose=True,
)

보안 고려사항

메모리 저장소를 구성할 때:

• 저장 경로에 환경 변수를 사용하세요 (예: CREWAI_STORAGE_DIR)
• 데이터베이스 자격 증명과 같은 민감 정보를 코드에 하드코딩하지 마세요
• 저장 디렉토리에 대한 접근 권한을 고려하세요
• 이동성을 위해 상대 경로를 사용하는 것이 좋습니다

환경 변수를 사용하는 예시:

import os
from crewai import Crew
from crewai.memory import LongTermMemory
from crewai.memory.storage.ltm_sqlite_storage import LTMSQLiteStorage

# 환경 변수를 사용한 저장소 경로 설정
storage_path = os.getenv("CREWAI_STORAGE_DIR", "./storage")
crew = Crew(
    memory=True,
    long_term_memory=LongTermMemory(
        storage=LTMSQLiteStorage(
            db_path="{storage_path}/memory.db".format(storage_path=storage_path)
        )
    )
)

구성 예시

기본 메모리 구성

from crewai import Crew
from crewai.memory import LongTermMemory

# 기본 저장소 경로를 사용하는 간단한 메모리 구성
crew = Crew(memory=True)

memory=True로 설정하면 External Memory는 정의되지 않습니다. 어떤 외부 메모리가 적합한지 추론할 수 없기 때문입니다.

사용자 정의 저장소 구성

from crewai import Crew
from crewai.memory import LongTermMemory
from crewai.memory.storage.ltm_sqlite_storage import LTMSQLiteStorage

# 사용자 정의 저장 경로 설정
crew = Crew(
    memory=True,
    long_term_memory=LongTermMemory(
        storage=LTMSQLiteStorage(db_path="./memory.db")
    )
)

Mem0 통합을 통한 사용자 메모리 향상

Mem0는 LLM 애플리케이션을 위한 자기개선형 메모리 레이어로, 개인화된 AI 경험을 가능하게 합니다.

Mem0 API 플랫폼 사용

사용자 맞춤 메모리를 포함시키려면 여기서 API 키를 발급받고, 공식 문서를 참고하여 사용자 선호 정보를 추가하세요. 이 경우 user_memory는 mem0의 MemoryClient로 설정됩니다.

import os
from crewai import Crew, Process
from mem0 import MemoryClient

# Mem0를 위한 환경 변수 설정
os.environ["MEM0_API_KEY"] = "m0-xx"

# 1단계: 사용자 메모리를 포함한 Crew 생성
crew = Crew(
    agents=[...],
    tasks=[...],
    verbose=True,
    process=Process.sequential,
    memory=True,
    memory_config={
        "provider": "mem0",
        "config": {"user_id": "john"},
        "user_memory": {} # 현재 이 이슈로 인해 명시적으로 dictionary로 설정 필요
    },
)

추가 메모리 구성 옵션

특정 조직 및 프로젝트에 접근하려면 org_id 및 project_id 매개변수를 설정할 수 있습니다.

from crewai import Crew

crew = Crew(
    agents=[...],
    tasks=[...],
    verbose=True,
    memory=True,
    memory_config={
        "provider": "mem0",
        "config": {"user_id": "john", "org_id": "my_org_id", "project_id": "my_project_id"},
        "user_memory": {} # 현재 이 이슈로 인해 명시적으로 dictionary로 설정 필요
    },
)

로컬 Mem0 메모리 사용하기

로컬 Mem0 메모리를 사용자 정의 구성으로 사용하고 싶다면, 구성 내에 local_mem0_config 파라미터를 설정할 수 있습니다. os 환경 변수와 local_mem0_config가 모두 설정된 경우, Mem0 API 플랫폼이 로컬 구성보다 우선됩니다. 더 자세한 내용은 이 문서를 참고하세요. 이 경우 user_memory는 mem0의 Memory 객체로 설정됩니다.

from crewai import Crew

# 로컬 Mem0 구성
config = {
    "vector_store": {
        "provider": "qdrant",
        "config": {
            "host": "localhost",
            "port": 6333
        }
    },
    "llm": {
        "provider": "openai",
        "config": {
            "api_key": "your-api-key",
            "model": "gpt-4"
        }
    },
    "embedder": {
        "provider": "openai",
        "config": {
            "api_key": "your-api-key",
            "model": "text-embedding-3-small"
        }
    },
    "graph_store": {
        "provider": "neo4j",
        "config": {
            "url": "neo4j+s://your-instance",
            "username": "neo4j",
            "password": "password"
        }
    },
    "history_db_path": "/path/to/history.db",
    "version": "v1.1",
    "custom_fact_extraction_prompt": "메모리를 위한 사실 추출용 커스텀 프롬프트 (선택사항)",
    "custom_update_memory_prompt": "메모리 업데이트용 커스텀 프롬프트 (선택사항)"
}

crew = Crew(
    agents=[...],
    tasks=[...],
    verbose=True,
    memory=True,
    memory_config={
        "provider": "mem0",
        "config": {"user_id": "john", 'local_mem0_config': config},
        "user_memory" : {} # 현재 이 이슈로 인해 명시적으로 dictionary로 설정 필요
    },
)

외부 메모리 사용하기

외부 메모리는 CrewAI 애플리케이션에 외부 메모리 시스템을 통합할 수 있는 강력한 기능입니다. 전문 메모리 제공자를 사용하거나 여러 애플리케이션 간 메모리를 공유하고 싶을 때 특히 유용합니다. 외부 메모리는 Long Term, Short Term 메모리와 달리 기본값을 설정할 수 없습니다.

Mem0와 함께 외부 메모리 사용 예시

import os
from crewai import Agent, Crew, Process, Task
from crewai.memory.external.external_memory import ExternalMemory

os.environ["MEM0_API_KEY"] = "YOUR-API-KEY"

agent = Agent(
    role="당신은 도움이 되는 어시스턴트입니다",
    goal="사용자의 휴가 계획을 세우세요",
    backstory="당신은 사용자의 휴가를 계획해주는 친절한 어시스턴트입니다",
    verbose=True,
)
task = Task(
    description="사용자의 휴가에 관련된 내용을 제공합니다",
    expected_output="휴가 계획",
    agent=agent,
)

crew = Crew(
    agents=[agent],
    tasks=[task],
    verbose=True,
    process=Process.sequential,
    external_memory=ExternalMemory(
        embedder_config={"provider": "mem0", "config": {"user_id": "U-123"}} # 전체 Mem0 구성 제공 가능
    ),
)

crew.kickoff(
    inputs={"question": "해변 휴가에 더 좋은 여행지는 어디인가요?"}
)

커스텀 저장소를 사용한 외부 메모리

외부 메모리에 대한 커스텀 저장소 구현도 가능합니다. 다음은 커스텀 저장소 예시입니다:

from crewai import Agent, Crew, Process, Task
from crewai.memory.external.external_memory import ExternalMemory
from crewai.memory.storage.interface import Storage

class CustomStorage(Storage):
    def __init__(self):
        self.memories = []

    def save(self, value, metadata=None, agent=None):
        self.memories.append({"value": value, "metadata": metadata, "agent": agent})

    def search(self, query, limit=10, score_threshold=0.5):
        # 여기에 검색 로직을 구현하세요
        return []

    def reset(self):
        self.memories = []

# 커스텀 저장소로 외부 메모리 생성
external_memory = ExternalMemory(
    storage=CustomStorage(),
    embedder_config={"provider": "mem0", "config": {"user_id": "U-123"}},
)

agent = Agent(
    role="당신은 도움이 되는 어시스턴트입니다",
    goal="사용자의 휴가 계획을 세우세요",
    backstory="당신은 사용자의 휴가를 계획해주는 친절한 어시스턴트입니다",
    verbose=True,
)
task = Task(
    description="사용자의 휴가에 관련된 내용을 제공합니다",
    expected_output="휴가 계획",
    agent=agent,
)

crew = Crew(
    agents=[agent],
    tasks=[task],
    verbose=True,
    process=Process.sequential,
    external_memory=external_memory,
)

crew.kickoff(
    inputs={"question": "해변 휴가에 더 좋은 여행지는 어디인가요?"}
)

추가 임베딩 제공자

OpenAI 임베딩 사용 (기본값)

from crewai import Crew, Agent, Task, Process

my_crew = Crew(
    agents=[...],
    tasks=[...],
    process=Process.sequential,
    memory=True,
    verbose=True,
    embedder={
        "provider": "openai",
        "config": {
            "model": 'text-embedding-3-small'
        }
    }
)

또는 OpenAIEmbeddingFunction을 직접 embedder 파라미터에 전달할 수도 있습니다.

예시:

from crewai import Crew, Agent, Task, Process
from chromadb.utils.embedding_functions import OpenAIEmbeddingFunction

my_crew = Crew(
    agents=[...],
    tasks=[...],
    process=Process.sequential,
    memory=True,
    verbose=True,
    embedder={
        "provider": "openai",
        "config": {
            "model": 'text-embedding-3-small'
        }
    }
)

Ollama 임베딩 사용

from crewai import Crew, Agent, Task, Process

my_crew = Crew(
    agents=[...],
    tasks=[...],
    process=Process.sequential,
    memory=True,
    verbose=True,
    embedder={
        "provider": "ollama",
        "config": {
            "model": "mxbai-embed-large"
        }
    }
)

Google AI 임베딩 사용

사전 준비

Google AI 임베딩을 사용하기 전에 다음이 필요합니다:

• Gemini API에 대한 접근 권한
• 필요한 API 키 및 권한

pyproject.toml 파일의 dependencies를 다음과 같이 업데이트해야 합니다:

dependencies = [
    "google-generativeai>=0.8.4", # 2025년 1월 기준 주요 버전 - crewai v.0.100.0 및 crewai-tools 0.33.0
    "crewai[tools]>=0.100.0,<1.0.0"
]

from crewai import Crew, Agent, Task, Process

my_crew = Crew(
    agents=[...],
    tasks=[...],
    process=Process.sequential,
    memory=True,
    verbose=True,
    embedder={
        "provider": "google",
        "config": {
            "api_key": "<YOUR_API_KEY>",
            "model": "<model_name>"
        }
    }
)

Azure OpenAI 임베딩 사용

from chromadb.utils.embedding_functions import OpenAIEmbeddingFunction
from crewai import Crew, Agent, Task, Process

my_crew = Crew(
    agents=[...],
    tasks=[...],
    process=Process.sequential,
    memory=True,
    verbose=True,
    embedder={
        "provider": "openai",
        "config": {
            "api_key": "YOUR_API_KEY",
            "api_base": "YOUR_API_BASE_PATH",
            "api_version": "YOUR_API_VERSION",
            "model_name": 'text-embedding-3-small'
        }
    }
)

Vertex AI 임베딩 사용

from chromadb.utils.embedding_functions import GoogleVertexEmbeddingFunction
from crewai import Crew, Agent, Task, Process

my_crew = Crew(
    agents=[...],
    tasks=[...],
    process=Process.sequential,
    memory=True,
    verbose=True,
    embedder={
        "provider": "vertexai",
        "config": {
            "project_id": "YOUR_PROJECT_ID",
            "region": "YOUR_REGION",
            "api_key": "YOUR_API_KEY",
            "model_name": "textembedding-gecko"
        }
    }
)

Cohere 임베딩 사용

from crewai import Crew, Agent, Task, Process

my_crew = Crew(
    agents=[...],
    tasks=[...],
    process=Process.sequential,
    memory=True,
    verbose=True,
    embedder={
        "provider": "cohere",
        "config": {
            "api_key": "YOUR_API_KEY",
            "model": "<model_name>"
        }
    }
)

VoyageAI 임베딩 사용

from crewai import Crew, Agent, Task, Process

my_crew = Crew(
    agents=[...],
    tasks=[...],
    process=Process.sequential,
    memory=True,
    verbose=True,
    embedder={
        "provider": "voyageai",
        "config": {
            "api_key": "YOUR_API_KEY",
            "model": "<model_name>"
        }
    }
)

HuggingFace 임베딩 사용

from crewai import Crew, Agent, Task, Process

my_crew = Crew(
    agents=[...],
    tasks=[...],
    process=Process.sequential,
    memory=True,
    verbose=True,
    embedder={
        "provider": "huggingface",
        "config": {
            "api_url": "<api_url>"
        }
    }
)

Watson 임베딩 사용

from crewai import Crew, Agent, Task, Process

# 주의: Watson 임베딩을 사용하려면 `ibm_watsonx_ai` 패키지가 설치 및 임포트되어 있어야 합니다.

my_crew = Crew(
    agents=[...],
    tasks=[...],
    process=Process.sequential,
    memory=True,
    verbose=True,
    embedder={
        "provider": "watson",
        "config": {
            "model": "<model_name>",
            "api_url": "<api_url>",
            "api_key": "<YOUR_API_KEY>",
            "project_id": "<YOUR_PROJECT_ID>"
        }
    }
)

Amazon Bedrock 임베딩 사용

# 주의: Bedrock 임베딩을 사용하려면 `boto3` 패키지가 설치되어 있어야 합니다.

import os
import boto3
from crewai import Crew, Agent, Task, Process

boto3_session = boto3.Session(
    region_name=os.environ.get("AWS_REGION_NAME"),
    aws_access_key_id=os.environ.get("AWS_ACCESS_KEY_ID"),
    aws_secret_access_key=os.environ.get("AWS_SECRET_ACCESS_KEY")
)

my_crew = Crew(
    agents=[...],
    tasks=[...],
    process=Process.sequential,
    memory=True,
    embedder={
        "provider": "bedrock",
        "config": {
            "session": boto3_session,
            "model": "amazon.titan-embed-text-v2:0",
            "vector_dimension": 1024
        }
    },
    verbose=True
)

커스텀 임베딩 함수 추가하기

from crewai import Crew, Agent, Task, Process
from chromadb import Documents, EmbeddingFunction, Embeddings

# 커스텀 임베딩 함수 생성
class CustomEmbedder(EmbeddingFunction):
    def __call__(self, input: Documents) -> Embeddings:
        # 임베딩 생성
        return [1, 2, 3] # 더미 임베딩 예시

my_crew = Crew(
    agents=[...],
    tasks=[...],
    process=Process.sequential,
    memory=True,
    verbose=True,
    embedder={
        "provider": "custom",
        "config": {
            "embedder": CustomEmbedder()
        }
    }
)

CLI로 메모리 초기화하기

crewai reset-memories [OPTIONS]

메모리 초기화 옵션

옵션	설명	타입	기본값
-l, --long	장기 메모리 초기화	플래그 (boolean)	False
-s, --short	단기 메모리 초기화	플래그 (boolean)	False
-e, --entities	엔티티 메모리 초기화	플래그 (boolean)	False
-k, --kickoff-outputs	최근 kickoff 태스크 결과 초기화	플래그 (boolean)	False
-kn, --knowledge	지식 저장소 초기화	플래그 (boolean)	False
-a, --all	모든 메모리 초기화	플래그 (boolean)	False

※ CLI 명령어를 사용하려면 같은 디렉토리에 crew.py 파일이 존재해야 합니다.

Crew 객체를 통한 메모리 초기화

my_crew = Crew(
    agents=[...],
    tasks=[...],
    process=Process.sequential,
    memory=True,
    verbose=True,
    embedder={
        "provider": "custom",
        "config": {
            "embedder": CustomEmbedder()
        }
    }
)

my_crew.reset_memories(command_type='all') # 모든 메모리 초기화

메모리 초기화 명령어 종류

명령어 타입	설명
long	장기 메모리 초기화
short	단기 메모리 초기화
entities	엔티티 메모리 초기화
kickoff_outputs	최근 kickoff 태스크 결과 초기화
knowledge	지식 메모리 초기화
all	모든 메모리 초기화

CrewAI 메모리 시스템의 장점

• 적응형 학습: 크루는 새로운 정보에 적응하며 시간이 지날수록 더 효율적으로 작업합니다.
• 개인화 향상: 에이전트는 사용자 선호 및 과거 상호작용을 기억하여 맞춤형 경험을 제공합니다.
• 문제 해결 능력 향상: 풍부한 메모리 접근을 통해 과거 학습 및 컨텍스트 인사이트를 기반으로 더 나은 결정을 내릴 수 있습니다.

결론

CrewAI의 메모리 시스템을 프로젝트에 통합하는 것은 매우 간단합니다. 제공되는 메모리 구성 요소 및 설정을 활용하면, 에이전트가 기억하고 추론하며 학습할 수 있는 능력을 빠르게 부여할 수 있어, 더 높은 수준의 지능과 기능을 발휘할 수 있습니다.