CrewAI 연동

CrewAI 연동 (crewai)

CrewAI는 역할극을 하는(role-playing) 자율 AI 에이전트를 오케스트레이션하는 프레임워크예요. 협업 지능(collaborative intelligence)을 활용해서 여러 에이전트가 매끄럽게 함께 작업하며 복잡한 태스크를 처리하도록 해줘요.

이 프레임워크는 AI 에이전트의 역량을 크게 끌어올리는 정교한 메모리 시스템을 갖추고 있어요. 이 시스템은 에이전트가 과거 상호작용을 기억하고, 추론하며, 학습하도록 도와줘요. Qdrant로 CrewAI 에이전트의 **단기 메모리(short-term memory)**와 **엔티티 메모리(entity memory)**를 저장할 수 있어요.

단기 메모리 (Short-Term Memory) — RAG를 사용해 최근 상호작용과 결과를 임시로 저장해요. 에이전트가 현재 실행 중에 자신의 현재 맥락과 관련된 정보를 떠올리고 활용할 수 있게 해줘요.
엔티티 메모리 (Entity Memory) — 태스크 중 만난 엔티티(사람, 장소, 개념)에 대한 정보를 포착하고 정리해요. 더 깊은 이해와 관계 매핑을 돕고, 엔티티 정보 저장에 RAG를 사용해요.

출처: Qdrant 공식 문서 — crewai

Qdrant와 함께 쓰기 (Usage with Qdrant)

CrewAI의 기본 메모리 저장소를 Qdrant를 쓰도록 커스터마이징하는 방법을 배워볼게요.

설치 (Installation)

먼저 CrewAI와 Qdrant 클라이언트 패키지를 설치해요.

pip install 'crewai[tools]' 'qdrant-client[fastembed]'

CrewAI 프로젝트 설정하기 (Setup a CrewAI Project)

CrewAI 프로젝트를 만드는 방법은 여기에서 배울 수 있어요. 프로젝트 이름이 mycrew라고 가정할게요.

Qdrant 스토리지 정의하기 (Define the Qdrant storage)

src/mycrew/storage.py

from typing import Any, Dict, List, Optional

from crewai.memory.storage.rag_storage import RAGStorage
from qdrant_client import QdrantClient


class QdrantStorage(RAGStorage):
    """
    Extends Storage to handle embeddings for memory entries using Qdrant.

    """

    def __init__(self, type, allow_reset=True, embedder_config=None, crew=None):
        super().__init__(type, allow_reset, embedder_config, crew)

    def search(
        self,
        query: str,
        limit: int = 3,
        filter: Optional[dict] = None,
        score_threshold: float = 0,
    ) -> List[Any]:
        points = self.client.query(
            self.type,
            query_text=query,
            query_filter=filter,
            limit=limit,
            score_threshold=score_threshold,
        )
        results = [
            {
                "id": point.id,
                "metadata": point.metadata,
                "context": point.document,
                "score": point.score,
            }
            for point in points
        ]

        return results

    def reset(self) -> None:
        self.client.delete_collection(self.type)

    def _initialize_app(self):
        self.client = QdrantClient()
        # uncomment the next line of code
        # and choose from the [supported embedders](https://qdrant.github.io/fastembed/examples/Supported_Models/)
        # if you don't want to use the default one
        # self.client._embedding_model_name = 'jinaai/jina-embeddings-v2-small-en'
        if not self.client.collection_exists(self.type):
            self.client.create_collection(
                collection_name=self.type,
                vectors_config=self.client.get_fastembed_vector_params(),
                sparse_vectors_config=self.client.get_fastembed_sparse_vector_params(),
            )

    def save(self, value: Any, metadata: Dict[str, Any]) -> None:
        self.client.add(self.type, documents=[value], metadata=[metadata or {}])

add와 query 메서드는 FastEmbed를 사용해 데이터를 벡터화해요. 필요하면 이 임베딩 모델을 커스터마이징할 수도 있어요.

크루(Crew) 인스턴스화하기 (Instantiate your crew)

Crew를 위한 에이전트와 태스크를 설정하는 방법은 여기에서 배울 수 있어요. Crew의 인스턴스화를 우리의 저장 메커니즘을 쓰도록 바꿔볼게요.

src/mycrew/crew.py

from crewai import Crew
from crewai.memory.entity.entity_memory import EntityMemory
from crewai.memory.short_term.short_term_memory import ShortTermMemory

from mycrew.storage import QdrantStorage

Crew(
    # Import the agents and tasks here.
    memory=True,
    entity_memory=EntityMemory(storage=QdrantStorage("entity")),
    short_term_memory=ShortTermMemory(storage=QdrantStorage("short-term")),
)

이제 crew run으로 Crew 워크플로우를 실행할 수 있어요. 메모리의 저장(ingestion)과 검색(retrieval)에 Qdrant를 사용해요.