메모리와 RAG

메모리와 RAG

에이전트를 만들다 보면 "유용한 사실들의 저장소"를 유지하고 싶을 때가 많아요. 특정 단계 직전에 그 사실들을 에이전트의 컨텍스트에 지능적으로 추가하는 것이죠. 대표적인 사례가 RAG 패턴이에요. 쿼리로 데이터베이스에서 관련 정보를 검색해 에이전트의 컨텍스트에 더하는 방식이죠. 이번에는 AgentChat이 제공하는 Memory 프로토콜을 다양한 저장소(간단한 리스트부터 벡터 DB, Redis까지)로 활용하는 방법과, 이것을 묶어 완전한 RAG 에이전트를 만드는 방법을 살펴볼게요.

출처: 공식문서

Memory 프로토콜

AgentChat은 autogen_core.memory.Memory 프로토콜을 제공하는데, 이 프로토콜을 확장해 메모리 기능을 구현할 수 있어요. 핵심 메서드는 query, update_context, add, clear, close입니다.

• add: 메모리 저장소에 새 항목 추가
• query: 메모리 저장소에서 관련 정보 검색
• update_context: 검색한 정보를 추가해 에이전트의 내부 model_context를 변경(AssistantAgent 클래스에서 사용)
• clear: 메모리 저장소의 모든 항목 비우기
• close: 메모리 저장소가 쓰는 리소스 정리

ListMemory 예제

autogen_core.memory.ListMemory는 Memory 프로토콜의 예제 구현으로 제공돼요. 시계열 순서로 메모리를 유지하고, 가장 최근 메모리를 모델 컨텍스트에 추가하는 단순한 리스트 기반 구현이죠. 이해하고 디버깅하기 쉽도록 단순하고 예측 가능하게 설계돼 있어요. 아래 예제에서는 ListMemory로 사용자 선호도 메모리 뱅크를 유지하고, 시간이 지나도 에이전트 응답에 일관된 컨텍스트를 제공하는 모습을 보여줄게요.

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.ui import Console
from autogen_core.memory import ListMemory, MemoryContent, MemoryMimeType
from autogen_ext.models.openai import OpenAIChatCompletionClient

# Initialize user memory
user_memory = ListMemory()

# Add user preferences to memory
await user_memory.add(MemoryContent(content="The weather should be in metric units", mime_type=MemoryMimeType.TEXT))

await user_memory.add(MemoryContent(content="Meal recipe must be vegan", mime_type=MemoryMimeType.TEXT))


async def get_weather(city: str, units: str = "imperial") -> str:
    if units == "imperial":
        return f"The weather in {city} is 73 °F and Sunny."
    elif units == "metric":
        return f"The weather in {city} is 23 °C and Sunny."
    else:
        return f"Sorry, I don't know the weather in {city}."


assistant_agent = AssistantAgent(
    name="assistant_agent",
    model_client=OpenAIChatCompletionClient(
        model="gpt-4o-2024-08-06",
    ),
    tools=[get_weather],
    memory=[user_memory],
)

# Run the agent with a task.
stream = assistant_agent.run_stream(task="What is the weather in New York?")
await Console(stream)

이제 assistant_agent의 model_context가 실제로 검색된 메모리 항목으로 업데이트됐는지 확인할 수 있어요. transform 메서드는 검색된 메모리 항목을 에이전트가 사용할 수 있는 문자열로 포맷하는 데 쓰여요. 여기서는 각 메모리 항목의 내용을 단순히 하나의 문자열로 이어 붙였죠.

await assistant_agent._model_context.get_messages()

위에서 날씨가 사용자 선호도에 명시된 대로 섭씨(Centigrade)로 반환되는 걸 볼 수 있어요. 마찬가지로 식단에 관한 별도의 질문을 하면, 에이전트가 메모리 저장소에서 관련 정보를 검색해 개인화된(비건) 응답을 제공합니다.

stream = assistant_agent.run_stream(task="Write brief meal recipe with broth")
await Console(stream)

커스텀 메모리 저장소 (벡터 DB 등)

Memory 프로토콜을 확장하면 더 복잡한 메모리 저장소를 구현할 수 있어요. 예를 들어 벡터 데이터베이스를 사용해 정보를 저장·검색하는 커스텀 메모리 저장소, 또는 머신러닝 모델로 사용자 선호도에 기반한 개인화 응답을 생성하는 저장소를 만들 수 있죠. 구체적으로는 add, query, update_context 메서드를 오버로드해 원하는 기능을 구현하고, 그 메모리 저장소를 에이전트에 넘겨주면 됩니다.

현재 autogen_ext 확장 패키지에는 다음과 같은 예제 메모리 저장소가 있어요.

• autogen_ext.memory.chromadb.ChromaDBVectorMemory: 벡터 데이터베이스로 정보를 저장·검색하는 메모리 저장소.
• autogen_ext.memory.chromadb.SentenceTransformerEmbeddingFunctionConfig: ChromaDBVectorMemory가 사용하는 SentenceTransformer 임베딩 함수의 설정 클래스. autogen_ext.memory.openai.OpenAIEmbeddingFunctionConfig 같은 다른 임베딩 함수도 쓸 수 있어요.
• autogen_ext.memory.redis.RedisMemory: Redis 벡터 데이터베이스로 정보를 저장·검색하는 메모리 저장소.

import tempfile

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.ui import Console
from autogen_core.memory import MemoryContent, MemoryMimeType
from autogen_ext.memory.chromadb import (
    ChromaDBVectorMemory,
    PersistentChromaDBVectorMemoryConfig,
    SentenceTransformerEmbeddingFunctionConfig,
)
from autogen_ext.models.openai import OpenAIChatCompletionClient

# Use a temporary directory for ChromaDB persistence
with tempfile.TemporaryDirectory() as tmpdir:
    chroma_user_memory = ChromaDBVectorMemory(
        config=PersistentChromaDBVectorMemoryConfig(
            collection_name="preferences",
            persistence_path=tmpdir,  # Use the temp directory here
            k=2,  # Return top k results
            score_threshold=0.4,  # Minimum similarity score
            embedding_function_config=SentenceTransformerEmbeddingFunctionConfig(
                model_name="all-MiniLM-L6-v2"  # Use default model for testing
            ),
        )
    )
    # Add user preferences to memory
    await chroma_user_memory.add(
        MemoryContent(
            content="The weather should be in metric units",
            mime_type=MemoryMimeType.TEXT,
            metadata={"category": "preferences", "type": "units"},
        )
    )

    await chroma_user_memory.add(
        MemoryContent(
            content="Meal recipe must be vegan",
            mime_type=MemoryMimeType.TEXT,
            metadata={"category": "preferences", "type": "dietary"},
        )
    )

    model_client = OpenAIChatCompletionClient(
        model="gpt-4o",
    )

    # Create assistant agent with ChromaDB memory
    assistant_agent = AssistantAgent(
        name="assistant_agent",
        model_client=model_client,
        tools=[get_weather],
        memory=[chroma_user_memory],
    )

    stream = assistant_agent.run_stream(task="What is the weather in New York?")
    await Console(stream)

    await model_client.close()
    await chroma_user_memory.close()

참고로 ChromaDBVectorMemory를 직렬화해 디스크에 저장할 수도 있어요.

chroma_user_memory.dump_component().model_dump_json()

Redis 메모리

Redis로도 동일한 영속 메모리 저장을 할 수 있어요. 연결할 Redis 인스턴스가 실행 중이어야 합니다. 로컬이나 Docker로 Redis를 실행하는 방법은 autogen_ext.memory.redis.RedisMemory를 참고하세요.

from logging import WARNING, getLogger

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.ui import Console
from autogen_core.memory import MemoryContent, MemoryMimeType
from autogen_ext.memory.redis import RedisMemory, RedisMemoryConfig
from autogen_ext.models.openai import OpenAIChatCompletionClient

logger = getLogger()
logger.setLevel(WARNING)

# Initailize Redis memory
redis_memory = RedisMemory(
    config=RedisMemoryConfig(
        redis_url="redis://localhost:6379",
        index_name="chat_history",
        prefix="memory",
    )
)

# Add user preferences to memory
await redis_memory.add(
    MemoryContent(
        content="The weather should be in metric units",
        mime_type=MemoryMimeType.TEXT,
        metadata={"category": "preferences", "type": "units"},
    )
)

await redis_memory.add(
    MemoryContent(
        content="Meal recipe must be vegan",
        mime_type=MemoryMimeType.TEXT,
        metadata={"category": "preferences", "type": "dietary"},
    )
)

model_client = OpenAIChatCompletionClient(
    model="gpt-4o",
)

# Create assistant agent with ChromaDB memory
assistant_agent = AssistantAgent(
    name="assistant_agent",
    model_client=model_client,
    tools=[get_weather],
    memory=[redis_memory],
)

stream = assistant_agent.run_stream(task="What is the weather in New York?")
await Console(stream)

await model_client.close()
await redis_memory.close()

RAG 에이전트: 이것저것 모아서 만들기

AI 시스템에서 흔한 RAG(Retrieval Augmented Generation) 패턴은 두 가지 별개의 단계로 이뤄져요.

1. 인덱싱(Indexing): 문서를 로드하고, 청크로 나누고, 벡터 데이터베이스에 저장
2. 검색(Retrieval): 대화 런타임 동안 관련 청크를 찾아 사용

앞선 예제에서는 메모리에 항목을 수동으로 추가해 에이전트에 넘겼어요. 실제로는 인덱싱 과정이 보통 자동화되고, 제품 문서·내부 파일·지식 베이스 같은 훨씬 큰 문서 소스를 기반으로 하죠.

Note: RAG 시스템의 품질은 청킹과 검색 과정(모델, 임베딩 등)의 품질에 달려 있어요. 최상의 결과를 얻으려면 더 고급 청킹·검색 모델로 실험해봐야 할 수 있어요.

간단한 RAG 에이전트 만들기

먼저 문서를 로드하고, 청크로 나누고, ChromaDBVectorMemory 메모리 저장소에 저장하는 간단한 문서 인덱서를 만들어볼게요.

import re
from typing import List

import aiofiles
import aiohttp
from autogen_core.memory import Memory, MemoryContent, MemoryMimeType


class SimpleDocumentIndexer:
    """Basic document indexer for AutoGen Memory."""

    def __init__(self, memory: Memory, chunk_size: int = 1500) -> None:
        self.memory = memory
        self.chunk_size = chunk_size

    async def _fetch_content(self, source: str) -> str:
        """Fetch content from URL or file."""
        if source.startswith(("http://", "https://")):
            async with aiohttp.ClientSession() as session:
                async with session.get(source) as response:
                    return await response.text()
        else:
            async with aiofiles.open(source, "r", encoding="utf-8") as f:
                return await f.read()

    def _strip_html(self, text: str) -> str:
        """Remove HTML tags and normalize whitespace."""
        text = re.sub(r"<[^>]*>", " ", text)
        text = re.sub(r"\s+", " ", text)
        return text.strip()

    def _split_text(self, text: str) -> List[str]:
        """Split text into fixed-size chunks."""
        chunks: list[str] = []
        # Just split text into fixed-size chunks
        for i in range(0, len(text), self.chunk_size):
            chunk = text[i : i + self.chunk_size]
            chunks.append(chunk.strip())
        return chunks

    async def index_documents(self, sources: List[str]) -> int:
        """Index documents into memory."""
        total_chunks = 0

        for source in sources:
            try:
                content = await self._fetch_content(source)

                # Strip HTML if content appears to be HTML
                if "<" in content and ">" in content:
                    content = self._strip_html(content)

                chunks = self._split_text(content)

                for i, chunk in enumerate(chunks):
                    await self.memory.add(
                        MemoryContent(
                            content=chunk, mime_type=MemoryMimeType.TEXT, metadata={"source": source, "chunk_index": i}
                        )
                    )

                total_chunks += len(chunks)

            except Exception as e:
                print(f"Error indexing {source}: {str(e)}")

        return total_chunks

이 인덱서를 ChromaDBVectorMemory와 함께 사용해 완전한 RAG 에이전트를 만들어볼게요.

import os
from pathlib import Path

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.ui import Console
from autogen_ext.memory.chromadb import ChromaDBVectorMemory, PersistentChromaDBVectorMemoryConfig
from autogen_ext.models.openai import OpenAIChatCompletionClient

# Initialize vector memory

rag_memory = ChromaDBVectorMemory(
    config=PersistentChromaDBVectorMemoryConfig(
        collection_name="autogen_docs",
        persistence_path=os.path.join(str(Path.home()), ".chromadb_autogen"),
        k=3,  # Return top 3 results
        score_threshold=0.4,  # Minimum similarity score
    )
)

await rag_memory.clear()  # Clear existing memory


# Index AutoGen documentation
async def index_autogen_docs() -> None:
    indexer = SimpleDocumentIndexer(memory=rag_memory)
    sources = [
        "https://raw.githubusercontent.com/microsoft/autogen/main/README.md",
        "https://microsoft.github.io/autogen/dev/user-guide/agentchat-user-guide/tutorial/agents.html",
        "https://microsoft.github.io/autogen/dev/user-guide/agentchat-user-guide/tutorial/teams.html",
        "https://microsoft.github.io/autogen/dev/user-guide/agentchat-user-guide/tutorial/termination.html",
    ]
    chunks: int = await indexer.index_documents(sources)
    print(f"Indexed {chunks} chunks from {len(sources)} AutoGen documents")


await index_autogen_docs()

# Create our RAG assistant agent
rag_assistant = AssistantAgent(
    name="rag_assistant", model_client=OpenAIChatCompletionClient(model="gpt-4o"), memory=[rag_memory]
)

# Ask questions about AutoGen
stream = rag_assistant.run_stream(task="What is AgentChat?")
await Console(stream)

# Remember to close the memory when done
await rag_memory.close()

이 구현은 AutoGen 문서를 기반으로 질문에 답하는 RAG 에이전트를 제공해요. 질문을 하면 Memory 시스템이 관련 청크를 검색해 컨텍스트에 추가하고, 에이전트가 정보에 근거한 응답을 생성할 수 있게 해주죠.

프로덕션 시스템에서는 다음을 고려해볼 수 있어요.

1. 더 정교한 청킹 전략 구현
2. 메타데이터 필터링 기능 추가
3. 검색 스코어링 커스터마이즈
4. 도메인에 맞게 임베딩 모델 최적화

Mem0Memory 예제

autogen_ext.memory.mem0.Mem0Memory는 Mem0.ai의 메모리 시스템과의 통합을 제공해요. 클라우드 기반과 로컬 백엔드를 모두 지원해 에이전트에 고급 메모리 기능을 제공하죠. 구현이 적절한 검색과 컨텍스트 업데이트를 처리하기 때문에 프로덕션 환경에 적합해요.

다음 예제에서는 Mem0Memory로 대화를 넘나드는 영속 메모리를 유지하는 방법을 보여줄게요.

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.ui import Console
from autogen_core.memory import MemoryContent, MemoryMimeType
from autogen_ext.memory.mem0 import Mem0Memory
from autogen_ext.models.openai import OpenAIChatCompletionClient

# Initialize Mem0 cloud memory (requires API key)
# For local deployment, use is_cloud=False with appropriate config
mem0_memory = Mem0Memory(
    is_cloud=True,
    limit=5,  # Maximum number of memories to retrieve
)

# Add user preferences to memory
await mem0_memory.add(
    MemoryContent(
        content="The weather should be in metric units",
        mime_type=MemoryMimeType.TEXT,
        metadata={"category": "preferences", "type": "units"},
    )
)

await mem0_memory.add(
    MemoryContent(
        content="Meal recipe must be vegan",
        mime_type=MemoryMimeType.TEXT,
        metadata={"category": "preferences", "type": "dietary"},
    )
)

# Create assistant with mem0 memory
assistant_agent = AssistantAgent(
    name="assistant_agent",
    model_client=OpenAIChatCompletionClient(
        model="gpt-4o-2024-08-06",
    ),
    tools=[get_weather],
    memory=[mem0_memory],
)

# Ask about the weather
stream = assistant_agent.run_stream(task="What are my dietary preferences?")
await Console(stream)

위 예제는 Mem0Memory를 어시스턴트 에이전트와 함께 쓰는 방법을 보여줘요. 이 메모리 통합이 보장하는 것은 다음과 같아요.

1. 모든 에이전트 상호작용이 향후 참조를 위해 Mem0에 저장됨
2. 관련 메모리(사용자 선호도 같은 것)가 자동으로 검색되어 컨텍스트에 추가됨
3. 에이전트가 저장된 메모리에 기반해 일관된 행동을 유지할 수 있음

Mem0Memory는 특히 다음에 유용해요.

• 영속 메모리가 필요한 장기 실행 에이전트 배포
• 향상된 프라이버시 컨트롤이 필요한 애플리케이션
• 에이전트 전반의 통합 메모리 관리를 원하는 팀
• 고급 메모리 필터링과 분석이 필요한 사용 사례

ChromaDBVectorMemory처럼 Mem0Memory 설정도 직렬화할 수 있어요.

# Serialize the memory configuration
config_json = mem0_memory.dump_component().model_dump_json()
print(f"Memory config JSON: {config_json[:100]}...")

더 알아보기 (Learn more)

• 메모리가 적용된 어시스턴트 에이전트의 구조는 에이전트 튜토리얼을 참고하세요.
• 여러 에이전트를 팀으로 묶는 방법은 팀 튜토리얼을 확인하세요.
• RAG 검색의 기본 개념을 다지는 데는 벡터 데이터베이스에 대한 별도 문서가 도움이 돼요.